diff --git a/common/trans.c b/common/trans.c index 408b4a7e..ae150940 100644 --- a/common/trans.c +++ b/common/trans.c @@ -90,6 +90,86 @@ trans_get_wait_objs(struct trans *self, tbus *objs, int *count) return 0; } +/*****************************************************************************/ +int APP_CC +trans_get_wait_objs_rw(struct trans *self, + tbus *robjs, int *rcount, + tbus *wobjs, int *wcount) +{ + if (self == 0) + { + return 1; + } + + if (self->status != TRANS_STATUS_UP) + { + return 1; + } + + robjs[*rcount] = self->sck; + (*rcount)++; + + if (self->wait_s != 0) + { + wobjs[*wcount] = self->sck; + (*wcount)++; + } + + return 0; +} + +/*****************************************************************************/ +int APP_CC +send_waiting(struct trans *self, int block) +{ + struct stream *temp_s; + int bytes; + int sent; + int timeout; + int cont; + + timeout = block ? 100 : 0; + cont = 1; + while (cont) + { + if (self->wait_s != 0) + { + temp_s = self->wait_s; + if (g_tcp_can_send(self->sck, timeout)) + { + bytes = (int) (temp_s->end - temp_s->p); + sent = g_tcp_send(self->sck, temp_s->p, bytes, 0); + if (sent > 0) + { + temp_s->p += sent; + if (temp_s->p >= temp_s->end) + { + self->wait_s = (struct stream *) (temp_s->next_packet); + free_stream(temp_s); + } + } + else if (sent == 0) + { + return 1; + } + else + { + if (!g_tcp_last_error_would_block(self->sck)) + { + return 1; + } + } + } + } + else + { + break; + } + cont = block; + } + return 0; +} + /*****************************************************************************/ int APP_CC trans_check_wait_objs(struct trans *self) @@ -203,6 +283,12 @@ trans_check_wait_objs(struct trans *self) } } } + if (send_waiting(self, 0) != 0) + { + /* error */ + self->status = TRANS_STATUS_DOWN; + return 1; + } } return rv; @@ -291,7 +377,12 @@ trans_force_write_s(struct trans *self, struct stream *out_s) size = (int)(out_s->end - out_s->data); total = 0; - self->in_write = 1; + if (send_waiting(self, 1) != 0) + { + self->status = TRANS_STATUS_DOWN; + return 1; + } + while (total < size) { sent = g_tcp_send(self->sck, out_s->data + total, size - total, 0); @@ -309,7 +400,6 @@ trans_force_write_s(struct trans *self, struct stream *out_s) { /* term */ self->status = TRANS_STATUS_DOWN; - self->in_write = 0; return 1; } } @@ -319,7 +409,6 @@ trans_force_write_s(struct trans *self, struct stream *out_s) { /* error */ self->status = TRANS_STATUS_DOWN; - self->in_write = 0; return 1; } } @@ -327,7 +416,6 @@ trans_force_write_s(struct trans *self, struct stream *out_s) { /* error */ self->status = TRANS_STATUS_DOWN; - self->in_write = 0; return 1; } else @@ -335,7 +423,6 @@ trans_force_write_s(struct trans *self, struct stream *out_s) total = total + sent; } } - self->in_write = 0; return 0; } @@ -349,15 +436,12 @@ trans_force_write(struct trans *self) /*****************************************************************************/ int APP_CC -trans_write_check(struct trans* self, int timeout) +trans_write_copy(struct trans *self) { int size; - int total; - int sent; - int error; - tbus robjs[1]; - tbus wobjs[1]; struct stream *out_s; + struct stream *wait_s; + struct stream *temp_s; if (self->status != TRANS_STATUS_UP) { @@ -365,63 +449,24 @@ trans_write_check(struct trans* self, int timeout) } out_s = self->out_s; - size = (int)(out_s->end - out_s->data); - total = 0; - - self->in_write = 1; - while (total < size) + make_stream(wait_s); + init_stream(wait_s, size); + out_uint8a(wait_s, out_s->data, size); + s_mark_end(wait_s); + if (self->wait_s == 0) { - robjs[0] = self->sck; - wobjs[0] = self->sck; - error = g_obj_wait(robjs, 1, wobjs, 1, timeout); - if (error != 0) - { - /* error */ - self->status = TRANS_STATUS_DOWN; - self->in_write = 0; - return 1; - } - - if (!g_tcp_can_send(self->sck, 0)) - { - trans_check_wait_objs(self); - continue; - } - - sent = g_tcp_send(self->sck, out_s->data + total, size - total, 0); - - if (sent == -1) - { - if (g_tcp_last_error_would_block(self->sck)) - { - if (!g_tcp_can_send(self->sck, 10)) - { - /* check for term here */ - } - } - else - { - /* error */ - self->status = TRANS_STATUS_DOWN; - self->in_write = 0; - return 1; - } - } - else if (sent == 0) - { - /* error */ - self->status = TRANS_STATUS_DOWN; - self->in_write = 0; - return 1; - } - else + self->wait_s = wait_s; + } + else + { + temp_s = self->wait_s; + while (temp_s->next_packet != 0) { - total = total + sent; + temp_s = (struct stream *) (temp_s->next_packet); } + temp_s->next_packet = wait_s; } - self->in_write = 0; - return 0; } diff --git a/common/trans.h b/common/trans.h index d9ad6c19..b7b9c20d 100644 --- a/common/trans.h +++ b/common/trans.h @@ -54,7 +54,7 @@ struct trans struct stream* out_s; char* listen_filename; tis_term is_term; /* used to test for exit */ - int in_write; + struct stream* wait_s; }; struct trans* APP_CC @@ -74,7 +74,7 @@ trans_force_read(struct trans* self, int size); int APP_CC trans_force_write(struct trans* self); int APP_CC -trans_write_check(struct trans* self, int timeout); +trans_write_copy(struct trans* self); int APP_CC trans_connect(struct trans* self, const char* server, const char* port, int timeout); diff --git a/sesman/chansrv/chansrv.c b/sesman/chansrv/chansrv.c index 56626c77..86c50960 100644 --- a/sesman/chansrv/chansrv.c +++ b/sesman/chansrv/chansrv.c @@ -296,18 +296,8 @@ send_data_from_chan_item(struct chan_item *chan_item) LOGM((LOG_LEVEL_DEBUG, "chansrv::send_data_from_chan_item: -- " "size %d chan_flags 0x%8.8x", size, chan_flags)); g_sent = 1; - if (g_con_trans->in_write) - { - g_writeln("chansrv::send_data_from_chan_item: error, " - "write while in_write"); - error = 1; - } - else - { - /* write but check for read if blocked */ - error = trans_write_check(g_con_trans, -1); - } + error = trans_write_copy(g_con_trans); if (error != 0) { return 1; @@ -402,7 +392,7 @@ send_init_response_message(void) out_uint32_le(s, 2); /* msg id */ out_uint32_le(s, 8); /* size */ s_mark_end(s); - return trans_force_write(g_con_trans); + return trans_write_copy(g_con_trans); } /*****************************************************************************/ @@ -425,7 +415,7 @@ send_channel_setup_response_message(void) out_uint32_le(s, 4); /* msg id */ out_uint32_le(s, 8); /* size */ s_mark_end(s); - return trans_force_write(g_con_trans); + return trans_write_copy(g_con_trans); } /*****************************************************************************/ @@ -448,7 +438,7 @@ send_channel_data_response_message(void) out_uint32_le(s, 6); /* msg id */ out_uint32_le(s, 8); /* size */ s_mark_end(s); - return trans_force_write(g_con_trans); + return trans_write_copy(g_con_trans); } /*****************************************************************************/ @@ -647,7 +637,7 @@ process_message_channel_data(struct stream *s) if (chan_flags & 2) /* last */ { s_mark_end(ls); - trans_force_write(g_api_con_trans); + trans_write_copy(g_api_con_trans); } } } @@ -1051,7 +1041,9 @@ THREAD_RV THREAD_CC channel_thread_loop(void *in_val) { tbus objs[32]; + tbus wobjs[32]; int num_objs; + int num_wobjs; int timeout; int error; THREAD_RV rv; @@ -1065,12 +1057,13 @@ channel_thread_loop(void *in_val) { timeout = -1; num_objs = 0; + num_wobjs = 0; objs[num_objs] = g_term_event; num_objs++; trans_get_wait_objs(g_lis_trans, objs, &num_objs); trans_get_wait_objs(g_api_lis_trans, objs, &num_objs); - while (g_obj_wait(objs, num_objs, 0, 0, timeout) == 0) + while (g_obj_wait(objs, num_objs, wobjs, num_wobjs, timeout) == 0) { check_timeout(); if (g_is_wait_obj_set(g_term_event)) @@ -1145,10 +1138,12 @@ channel_thread_loop(void *in_val) xfuse_check_wait_objs(); timeout = -1; num_objs = 0; + num_wobjs = 0; objs[num_objs] = g_term_event; num_objs++; trans_get_wait_objs(g_lis_trans, objs, &num_objs); - trans_get_wait_objs(g_con_trans, objs, &num_objs); + trans_get_wait_objs_rw(g_con_trans, objs, &num_objs, + wobjs, &num_wobjs); trans_get_wait_objs(g_api_lis_trans, objs, &num_objs); trans_get_wait_objs(g_api_con_trans, objs, &num_objs); xcommon_get_wait_objs(objs, &num_objs, &timeout);