chansrv: deadlock, add trans copy func

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;
+    make_stream(wait_s);
-
+    init_stream(wait_s, size);
-    self->in_write = 1;
+    out_uint8a(wait_s, out_s->data, size);
-    while (total < size)
+    s_mark_end(wait_s);
+    if (self->wait_s == 0)
     {
-        robjs[0] = self->sck;
+        self->wait_s = wait_s;
-        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 */
+        temp_s = self->wait_s;
-                self->status = TRANS_STATUS_DOWN;
+        while (temp_s->next_packet != 0)
-                self->in_write = 0;
-                return 1;
-            }
-        }
-        else if (sent == 0)
         {
-            /* error */
+            temp_s = (struct stream *) (temp_s->next_packet);
-            self->status = TRANS_STATUS_DOWN;
-            self->in_write = 0;
-            return 1;
         }
-        else
+        temp_s->next_packet = wait_s;
-        {
-            total = total + sent;
     }
-    }
-    self->in_write = 0;
-
     return 0;
 }
 

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);

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);