43 	struct pipe *pp;  in mtcp_pipe()  local
58 	pp = (struct pipe *)calloc(1, sizeof(struct pipe));  in mtcp_pipe()
59 	if (!pp) {  in mtcp_pipe()
66 	pp->buf_size = PIPE_BUF_SIZE;  in mtcp_pipe()
67 	pp->buf = (char *)malloc(pp->buf_size);  in mtcp_pipe()
68 	if (!pp->buf) {  in mtcp_pipe()
72 		free(pp);  in mtcp_pipe()
76 	ret = pthread_mutex_init(&pp->pipe_lock, NULL);  in mtcp_pipe()
81 		free(pp->buf);  in mtcp_pipe()
82 		free(pp);  in mtcp_pipe()
86 	ret = pthread_cond_init(&pp->pipe_cond, NULL);  in mtcp_pipe()
91 		free(pp->buf);  in mtcp_pipe()
92 		pthread_mutex_destroy(&pp->pipe_lock);  in mtcp_pipe()
93 		free(pp);  in mtcp_pipe()
97 	pp->state = PIPE_ACTIVE;  in mtcp_pipe()
98 	pp->socket[0] = socket[0];  in mtcp_pipe()
99 	pp->socket[1] = socket[1];  in mtcp_pipe()
100 	socket[0]->pp = pp;  in mtcp_pipe()
101 	socket[1]->pp = pp;  in mtcp_pipe()
113 	struct pipe *pp = socket->pp;  in RaiseEventToPair()  local
116 	if (pp->socket[0] == socket)  in RaiseEventToPair()
117 		pair_socket = pp->socket[1];  in RaiseEventToPair()
119 		pair_socket = pp->socket[0];  in RaiseEventToPair()
126 		pthread_cond_signal(&pp->pipe_cond);  in RaiseEventToPair()
135 	struct pipe *pp;  in PipeRead()  local
152 	pp = socket->pp;  in PipeRead()
153 	if (!pp) {  in PipeRead()
157 	if (pp->state == PIPE_CLOSED) {  in PipeRead()
161 	if (pp->state == PIPE_CLOSE_WAIT && pp->buf_len == 0) {  in PipeRead()
174 	pthread_mutex_lock(&pp->pipe_lock);  in PipeRead()
176 		while (pp->buf_len == 0) {  in PipeRead()
177 			ret = pthread_cond_wait(&pp->pipe_cond, &pp->pipe_lock);  in PipeRead()
180 				pthread_mutex_unlock(&pp->pipe_lock);  in PipeRead()
186 	to_read = MIN(len, pp->buf_len);  in PipeRead()
188 		pthread_mutex_unlock(&pp->pipe_lock);  in PipeRead()
189 		if (pp->state == PIPE_ACTIVE) {  in PipeRead()
192 		} else if (pp->state == PIPE_CLOSE_WAIT) {  in PipeRead()
199 	if (pp->buf_len == pp->buf_size)  in PipeRead()
202 	if (pp->buf_off + to_read < pp->buf_size) {  in PipeRead()
203 		memcpy(buf, pp->buf + pp->buf_off, to_read);  in PipeRead()
204 		pp->buf_off += to_read;  in PipeRead()
206 		int temp_read = pp->buf_size - pp->buf_off;  in PipeRead()
207 		memcpy(buf, pp->buf + pp->buf_off, temp_read);  in PipeRead()
208 		memcpy(buf + temp_read, pp->buf, to_read - temp_read);  in PipeRead()
209 		pp->buf_off = to_read - temp_read;  in PipeRead()
211 	pp->buf_len -= to_read;  in PipeRead()
218 	pthread_mutex_unlock(&pp->pipe_lock);  in PipeRead()
221 	if (pp->buf_len > 0) {  in PipeRead()
226 	} else if (pp->state == PIPE_CLOSE_WAIT && pp->buf_len == 0) {  in PipeRead()
238 	struct pipe *pp;  in PipeWrite()  local
255 	pp = socket->pp;  in PipeWrite()
256 	if (!pp) {  in PipeWrite()
260 	if (pp->state == PIPE_CLOSED) {  in PipeWrite()
264 	if (pp->state == PIPE_CLOSE_WAIT) {  in PipeWrite()
278 	pthread_mutex_lock(&pp->pipe_lock);  in PipeWrite()
280 		while (pp->buf_len == pp->buf_size) {  in PipeWrite()
281 			ret = pthread_cond_wait(&pp->pipe_cond, &pp->pipe_lock);  in PipeWrite()
284 				pthread_mutex_unlock(&pp->pipe_lock);  in PipeWrite()
290 	to_write = MIN(len, pp->buf_size - pp->buf_len);  in PipeWrite()
292 		pthread_mutex_unlock(&pp->pipe_lock);  in PipeWrite()
299 	if (pp->buf_len == 0)  in PipeWrite()
302 	if (pp->buf_tail + to_write < pp->buf_size) {  in PipeWrite()
304 		memcpy(pp->buf + pp->buf_tail, buf, to_write);  in PipeWrite()
305 		pp->buf_tail += to_write;  in PipeWrite()
308 		int temp_write = pp->buf_size - pp->buf_tail;  in PipeWrite()
309 		memcpy(pp->buf + pp->buf_tail, buf, temp_write);  in PipeWrite()
310 		memcpy(pp->buf, buf + temp_write, to_write - temp_write);  in PipeWrite()
311 		pp->buf_tail = to_write - temp_write;  in PipeWrite()
313 	pp->buf_len += to_write;  in PipeWrite()
320 	pthread_mutex_unlock(&pp->pipe_lock);  in PipeWrite()
323 	if (pp->buf_len < pp->buf_size) {  in PipeWrite()
338 	struct pipe *pp = socket->pp;  in RaisePendingPipeEvents()  local
340 	if (!pp)  in RaisePendingPipeEvents()
342 	if (pp->state < PIPE_ACTIVE)  in RaisePendingPipeEvents()
348 		if (pp->buf_len > 0) {  in RaisePendingPipeEvents()
350 		} else if (pp->state == PIPE_CLOSE_WAIT) {  in RaisePendingPipeEvents()
357 		if (pp->buf_len < pp->buf_size) {  in RaisePendingPipeEvents()
370 	struct pipe *pp;  in PipeClose()  local
384 	pp = socket->pp;  in PipeClose()
385 	if (!pp) {  in PipeClose()
389 	if (pp->state == PIPE_CLOSED) {  in PipeClose()
393 	pthread_mutex_lock(&pp->pipe_lock);  in PipeClose()
394 	if (pp->state == PIPE_ACTIVE) {  in PipeClose()
395 		pp->state = PIPE_CLOSE_WAIT;  in PipeClose()
397 		pthread_mutex_unlock(&pp->pipe_lock);  in PipeClose()
403 	if (pp->socket[0])  in PipeClose()
404 		pp->socket[0]->pp = NULL;  in PipeClose()
405 	if (pp->socket[1])  in PipeClose()
406 		pp->socket[1]->pp = NULL;  in PipeClose()
408 	pthread_mutex_unlock(&pp->pipe_lock);  in PipeClose()
410 	pthread_mutex_destroy(&pp->pipe_lock);  in PipeClose()
411 	pthread_cond_destroy(&pp->pipe_cond);  in PipeClose()
413 	free(pp->buf);  in PipeClose()
415 	free(pp);  in PipeClose()