socket: client connections/sends; raddbg: move ipc sender mode to using socket layer

src/raddbg/raddbg_main.c

@@ -940,6 +940,10 @@ entry_point(CmdLine *cmd_line)
           {
             StringJoin join = {s(""), s("\0"), s("")};
             String8 msg = str8_list_join(scratch.arena, &rd_state->cmd_outputs, &join);
+            if(msg.size == 0)
+            {
+              msg = s("done");
+            }
             sock_send(icp_sock_session, ipc_protocol, ipc_endpoint, msg, now_time_us()+5000000);
           }
         }
@@ -952,6 +956,43 @@ entry_point(CmdLine *cmd_line)
     {
       Temp scratch = scratch_begin(0, 0);
       
+      //- rjf: create socket session
+      SOCK_Session ipc_sock_session = sock_session_open(0, 0);
+      
+      //- rjf: form target raddbg endpoint address
+      SOCK_Endpoint endpoint = {0};
+      {
+        endpoint.address_u8[0] = 127;
+        endpoint.address_u8[3] = 1;
+        endpoint.port = (U16)ipc_port;
+      }
+      
+      //- rjf: form message from command line inputs
+      String8List parts = {0};
+      {
+        for EachIndex(idx, cmd_line->argc-1)
+        {
+          str8_list_push(scratch.arena, &parts, str8_cstring(cmd_line->argv[idx+1]));
+        }
+      }
+      StringJoin join = {str8_lit(""), str8_lit(" "), str8_lit("")};
+      String8 msg = str8_list_join(scratch.arena, &parts, &join);
+      
+      //- rjf: send to server, get outputs
+      String8List outputs = {0};
+      if(sock_send(ipc_sock_session, SOCK_Protocol_TCP, endpoint, msg, now_time_us()+10000000))
+      {
+        SOCK_Protocol outputs_protocol = SOCK_Protocol_TCP;
+        SOCK_Endpoint outputs_endpoint = {0};
+        String8 output_data = {0};
+        if(sock_recv(scratch.arena, ipc_sock_session, &outputs_protocol, &outputs_endpoint, &output_data, now_time_us()+10000000))
+        {
+          U8 split_char = 0;
+          outputs = str8_split(scratch.arena, output_data, &split_char, 1, 0);
+        }
+      }
+      
+#if 0
       //- rjf: grab explicit PID argument
       U32 dst_pid = 0;
       if(cmd_line_has_argument(cmd_line, str8_lit("pid")))
@@ -1040,6 +1081,7 @@ entry_point(CmdLine *cmd_line)
           semaphore_drop(ipc_main2sender_lock_semaphore);
         }
       }
+#endif
       
       //- rjf: write outputs to stdout
       for(String8Node *n = outputs.first; n != 0; n = n->next)

src/win32/socket/win32_socket.c

@@ -76,6 +76,7 @@ w32_sock_listener_thread_entry_point(void *p)
         con->endpoint = endpoint;
         con->protocol = SOCK_Protocol_TCP;
         con->socket = new_socket;
+        MemoryZeroStruct(&con->recv_overlapped);
         DLLPushBack(slot->first, slot->last, con);
         buf.len = sizeof(con->recv_buffer);
         buf.buf = con->recv_buffer;
@@ -226,25 +227,126 @@ sock_async_tick(void)
   for(SendTask *t = first_tcp_send; t != 0; t = t->next)
   {
     W32_SOCK_Session *session = t->session;
+    
+    // rjf: unpack endpoint
     U64 hash = u64_hash_from_str8(str8_struct(&t->endpoint));
     U64 slot_idx = hash%session->connection_slots_count;
     W32_SOCK_ConnectionSlot *slot = &session->connection_slots[slot_idx];
     Stripe *stripe = stripe_from_slot_idx(&session->connection_stripes, slot_idx);
-    RWMutexScope(stripe->rw_mutex, 1)
+    
+    // rjf: get existing socket for this endpoint
+    SOCKET ep_socket = -1;
+    RWMutexScope(stripe->rw_mutex, 0)
     {
       for(W32_SOCK_Connection *c = slot->first; c != 0; c = c->next)
       {
         if(MemoryMatchStruct(&c->endpoint, &t->endpoint))
         {
-          if(send(c->socket, t->data.str, t->data.size, 0) == SOCKET_ERROR)
+          ep_socket = c->socket;
+          break;
+        }
+      }
+    }
+    
+    // rjf: didn't get a socket? -> open socket
+    if(ep_socket == -1) RWMutexScope(stripe->rw_mutex, 1)
+    {
+      // rjf: try to get socket again, now that we have the write lock
+      W32_SOCK_Connection *con = 0;
+      for(W32_SOCK_Connection *c = slot->first; c != 0; c = c->next)
+      {
+        if(MemoryMatchStruct(&c->endpoint, &t->endpoint))
+        {
+          con = c;
+          break;
+        }
+      }
+      
+      // rjf: no socket still? -> create
+      if(con == 0)
+      {
+        // rjf: convert endpoint -> sockaddr
+        struct sockaddr_storage endpoint_sockaddr = {0};
+        int endpoint_sockaddr_size = 0;
+        {
+          switch(t->endpoint.kind)
           {
-            int error = WSAGetLastError();
-            if(error == WSAECONNRESET)
+            default:{}break;
+            case SOCK_EndpointKind_IPv4:
             {
-              DLLRemove(slot->first, slot->last, c);
-              c->next = stripe->free;
-              stripe->free = c;
-            }
+              struct sockaddr_in *dst = (struct sockaddr_in *)(&endpoint_sockaddr);
+              endpoint_sockaddr_size = sizeof(*dst);
+              dst->sin_family = AF_INET;
+              dst->sin_port = htons(t->endpoint.port);
+              MemoryCopy(&dst->sin_addr, &t->endpoint.address_u32[0], sizeof(U32));
+            }break;
+            case SOCK_EndpointKind_IPv6:
+            {
+              struct sockaddr_in6 *dst = (struct sockaddr_in6 *)(&endpoint_sockaddr);
+              endpoint_sockaddr_size = sizeof(*dst);
+              dst->sin6_family = AF_INET6;
+              dst->sin6_port = htons(t->endpoint.port);
+              MemoryCopy(&dst->sin6_addr, &t->endpoint.address_u128[0], sizeof(U128));
+            }break;
+          }
+        }
+        
+        // rjf: create
+        SOCKET new_socket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+        connect(new_socket, (struct sockaddr *)&endpoint_sockaddr, endpoint_sockaddr_size);
+        
+        // rjf: store in cache
+        con = (W32_SOCK_Connection *)stripe->free;
+        if(con != 0)
+        {
+          stripe->free = con->next;
+        }
+        else
+        {
+          con = push_array(stripe->arena, W32_SOCK_Connection, 1);
+        }
+        con->endpoint = t->endpoint;
+        con->protocol = SOCK_Protocol_TCP;
+        con->socket = new_socket;
+        DLLPushBack(slot->first, slot->last, con);
+        
+        // rjf: associate this socket with iocp
+        CreateIoCompletionPort((HANDLE)new_socket, session->iocp, 0, 0);
+        MemoryZeroStruct(&con->recv_overlapped);
+        
+        // rjf: kick off receive on this socket
+        WSABUF buf = {sizeof(con->recv_buffer), con->recv_buffer};
+        DWORD flags = MSG_PUSH_IMMEDIATE;
+        WSARecv(con->socket, &buf, 1, &con->recv_size, &flags, &con->recv_overlapped, 0);
+      }
+      
+      // rjf: get socket from cache
+      ep_socket = con->socket;
+    }
+    
+    // rjf: got socket? -> send
+    B32 send_failed = 0;
+    if(ep_socket != -1 && send(ep_socket, t->data.str, t->data.size, 0) == SOCKET_ERROR)
+    {
+      int error = WSAGetLastError();
+      if(error == WSAECONNRESET)
+      {
+        send_failed = 1;
+      }
+    }
+    
+    // rjf: got a socket, but send failed? -> connection closed
+    if(ep_socket != -1 && send_failed)
+    {
+      RWMutexScope(stripe->rw_mutex, 1)
+      {
+        for(W32_SOCK_Connection *c = slot->first; c != 0; c = c->next)
+        {
+          if(MemoryMatchStruct(&c->endpoint, &t->endpoint))
+          {
+            DLLRemove(slot->first, slot->last, c);
+            c->next = stripe->free;
+            stripe->free = c;
           }
         }
       }