diff --git a/src/raddbg/raddbg_main.c b/src/raddbg/raddbg_main.c index 72c302d4..d8ddb79a 100644 --- a/src/raddbg/raddbg_main.c +++ b/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) diff --git a/src/win32/socket/win32_socket.c b/src/win32/socket/win32_socket.c index 8bcfcdb9..ecf69a04 100644 --- a/src/win32/socket/win32_socket.c +++ b/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; } } }