#include #include #include #include #include #include #include #ifndef WIN32 #include #include #include #include #else #define WINVER 0x0501 #include #include #include #endif #include #include extern unsigned long crc32buf(unsigned char *buf, size_t len); #define MAX_UNACK 102400 // window size #define MAX_PACKET 1024 /* message types */ #define MT_DATA 'D' #define MT_ACK 'A' #define MT_REQUEST 'R' #define MT_QUIT 'Q' #define MT_HELLO 'H' #define MT_NOOP 'N' #define MAXSEQ 0x80000000 #define SEND_FROM_END 1 #define SEND_FROM_BEGINNING 0 int socket_ops; int safer_write(int f, unsigned char* b, int n) { while(n) { int r; if (socket_ops) r = send(f, b, n, 0); else r = write(f, b, n); if(!r) return 0; if(r==-1) { if (errno==EINTR || errno==EAGAIN) continue; return -1; } n-=r; b+=r; } return 0; } int s = -1; int clients_server_socket = -1; int tcp_helper_socket = -1; int tcp_helper_server_socket = -1; unsigned char tcp_helper_debt_buf[2*MAX_PACKET]; int tcp_helper_debt_buf_offset=0; int tcp_helper_pending_packet_length = -1; int tcp_helper_pending_packet_length_first_byte = -1; int tcp_helper_packet_ready = 0; int tcp_helper_active = 0; unsigned char tcp_helper_send_debt[MAX_PACKET]; int tcp_helper_send_debt_len = 0; struct addrinfo *address = NULL; struct addrinfo *address_copy = NULL; struct addrinfo *commaddress = NULL; int unacknowledged_data = 0; unsigned char circular_buffer[MAX_UNACK * 2]; int startp=0; int endp=0; unsigned long seq = 0; unsigned long ack_seq = 0; unsigned long sent_seq = 0; unsigned long recv_seq = 0; int servermode; int readfd; int writefd; int commmode = 0; /* Listen or connect to other stream socket instead of using stdin/stdout */ int server_received_hello = 0; int short_timeout = 0; int stale_mode_message_already_printed=0; int tcp_helper_period = 0; int skipped_acks = 0; void nonblocking_socket(int s) { #ifndef WIN32 fcntl(s, F_SETFL, O_NONBLOCK); #else unsigned long flags = 1; ioctlsocket(s, FIONBIO, &flags); #endif } void close_tcp_helper() { tcp_helper_pending_packet_length=-1; tcp_helper_pending_packet_length_first_byte=-1; tcp_helper_debt_buf_offset=0; tcp_helper_active=0; close(tcp_helper_socket); tcp_helper_socket=-1; tcp_helper_send_debt_len=0; tcp_helper_packet_ready=0; } void open_tcp_helper() { close_tcp_helper(); tcp_helper_socket = socket(commaddress->ai_family, SOCK_STREAM, 0); nonblocking_socket(tcp_helper_socket); connect(tcp_helper_socket, address_copy->ai_addr, address_copy->ai_addrlen); } time_t last_time_tried_tcp_helper=-1; int try_tcp_helper=0; void maybe_connect_tcp_helper() { if(servermode) return; if(tcp_helper_active) return; if(try_tcp_helper) return; if(!tcp_helper_period) return; time_t t = time(NULL); if (last_time_tried_tcp_helper != -1) { if (t-last_time_tried_tcp_helper < tcp_helper_period) { return; // too soon } } last_time_tried_tcp_helper=t; fprintf(stderr, "Actomatically trying TCP helper\n"); try_tcp_helper = 1; } int sendto_s_or_tcp_helper(const unsigned char* buf, size_t len, int flags) { int ret = -1; if (tcp_helper_active && tcp_helper_socket!=-1) { unsigned char bu[MAX_PACKET+32]; bu[0] = len>>8; bu[1] = len&0xFF; memcpy(bu+2, buf, len); ret = send(tcp_helper_socket, bu, len+2, flags); if (ret == len+2) { return ret; } if (ret <= 1) { // Unlucky close_tcp_helper(); fprintf(stderr, "Short write to TCP helper\n"); } tcp_helper_send_debt_len = len+2-ret; memcpy(tcp_helper_send_debt, buf+len-tcp_helper_send_debt_len, tcp_helper_send_debt_len); return -1; } return sendto(s, buf, len, flags, address->ai_addr, address->ai_addrlen); } int send_data_i(unsigned char* data, size_t len, unsigned long seq){ unsigned char buf[MAX_PACKET+16]; buf[0]=MT_DATA; *(unsigned long*)(buf+1) = htonl(seq); memcpy(buf+5, data, len); *(unsigned long*)(buf+5+len) = htonl(crc32buf(buf, 5+len)); return sendto_s_or_tcp_helper(buf, 1+4+4+len, 0); } void send_data(int send_from_end) { int len_to_send = endp - startp; int correction = 0; if (len_to_send > MAX_PACKET) { correction = len_to_send - MAX_PACKET; len_to_send = MAX_PACKET; } if (len_to_send == MAX_PACKET) { maybe_connect_tcp_helper(); } int ret; if (!send_from_end) { ret = send_data_i(circular_buffer+startp, len_to_send, seq - correction); } else { ret = send_data_i(circular_buffer+endp-len_to_send, len_to_send, seq); } short_timeout=1; } void send_simple(char simple) { unsigned char buf[16]; buf[0]=simple; *(unsigned long*)(buf+1) = htonl(crc32buf(buf, 1)); sendto_s_or_tcp_helper(buf, 5, 0); } void send_ack() { unsigned char buf[9]; buf[0]=MT_ACK; *(unsigned long*)(buf+1) = htonl(recv_seq); *(unsigned long*)(buf+5) = htonl(crc32buf(buf, 5)); sendto(s, buf, 9, 0, address->ai_addr, address->ai_addrlen); } void handle_term(int p) { close_tcp_helper(); send_simple(MT_QUIT); exit(0); } void handle_usr1(int p) { close_tcp_helper(); send_data(SEND_FROM_BEGINNING); send_simple(MT_REQUEST); } void handle_usr2(int p) { open_tcp_helper(); } int main(int argc, char* argv[]) { #ifdef WIN32 WSADATA wsaData; int iResult = WSAStartup( MAKEWORD(2,2), &wsaData ); if ( iResult != NO_ERROR ) { fprintf(stderr,"Error at WSAStartup()\n"); exit(1); } _setmode(_fileno(stdin), _O_BINARY); _setmode(_fileno(stdout), _O_BINARY); #endif if (argc<2 || !strcmp(argv[1], "--help") || !strcmp(argv[1], "-?")) { fprintf (stderr, "udpmstr v0.1; implemented by Vitaly \"_Vi\" Shukela; 2013; license=MIT\n" "Usage:\n" "\tudpmstr {{-c|--client}|{-s|--server}} host port [{-c|-s} host2 port2]\n" "\tFor client, UDPMSTR_PERIOD and UDPMSTR_TCPHELPER_PERIOD\n" ); return 1; } #ifndef WIN32 { struct sigaction sa = {{&handle_term}}; sigaction(SIGTERM, &sa, NULL); sigaction(SIGINT, &sa, NULL); sa.sa_handler = &handle_usr1; sigaction(SIGUSR1, &sa, NULL); sa.sa_handler = &handle_usr2; sigaction(SIGUSR2, &sa, NULL); } #endif unacknowledged_data = 0; memset(circular_buffer, 0, sizeof circular_buffer); startp=0; endp=0; seq = 0; ack_seq = 0; recv_seq = 0; server_received_hello = 0; if (!strcmp(argv[1], "-s") || !strcmp(argv[1], "--server")) servermode=1; else if (!strcmp(argv[1], "-c") || !strcmp(argv[1], "--client")) servermode=0; else { fprintf(stderr, "The first command line parameter should be --server or --client\n"); return 2; } if (argc == 7) commmode=1; struct addrinfo hints; memset(&hints, 0, sizeof(struct addrinfo)); hints.ai_socktype = SOCK_DGRAM; /* Datagram socket */ hints.ai_flags = AI_PASSIVE; char* host = argv[2]; char* port = argv[3]; if (!strcmp(host, "NULL")) host=NULL; if (!strcmp(port, "NULL")) port=NULL; char* commhost = NULL; char* commport = NULL; int commlisten = 0; if (commmode) { commlisten = !(argv[4][1]=='c' || argv[4][2]=='c'); commhost = argv[5]; commport = argv[6]; if (!strcmp(commhost, "NULL")) commhost=NULL; if (!strcmp(commport, "NULL")) commport=NULL; } int gai_error; if ((gai_error=getaddrinfo(host, port, &hints, &address))) { fprintf(stderr, "getaddrinfo 1: %s\n",gai_strerror(gai_error)); return 4; } if (address->ai_next) { fprintf(stderr, "Warning: using only one of addresses retuned by getaddrinfo\n"); } if (!address) { fprintf(stderr, "getaddrinfo returned no addresses\n"); return 6; } getaddrinfo(host, port, &hints, &address_copy); if (commmode) { hints.ai_socktype = SOCK_STREAM; hints.ai_flags = servermode ? 0 : AI_PASSIVE; gai_error=getaddrinfo(commhost, commport, &hints, &commaddress); if (gai_error) { fprintf(stderr, "getaddrinfo 2: %s\n",gai_strerror(gai_error)); return 13; } if (commaddress->ai_next) { fprintf(stderr, "Warning: using only one of addresses retuned by getaddrinfo 2\n"); } if (!commaddress) { fprintf(stderr, "getaddrinfo 2 returned no addresses\n"); return 15; } } if (s==-1) { s = socket(address->ai_family, address->ai_socktype, address->ai_protocol); if (s == -1) { perror("socket"); return 7; } if (servermode) { { int one = 1; setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one); } if (bind(s, address->ai_addr, address->ai_addrlen)) { perror("bind"); return 5; } tcp_helper_server_socket = socket(address->ai_family, SOCK_STREAM, 0); { int one = 1; setsockopt(tcp_helper_server_socket, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one); } if (bind(tcp_helper_server_socket, address->ai_addr, address->ai_addrlen)) { close(tcp_helper_server_socket); tcp_helper_server_socket = -1; fprintf(stderr, "Can't bind TCP helper socket\n"); } listen(tcp_helper_server_socket, 1); nonblocking_socket(tcp_helper_server_socket); } } if (commmode) { readfd = -1; writefd = -1; socket_ops = 1; } else { readfd = 0; writefd = 1; socket_ops = 0; } if (commlisten) { clients_server_socket = socket(commaddress->ai_family, commaddress->ai_socktype, commaddress->ai_protocol); int one = 1; setsockopt(clients_server_socket, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one); if (bind(clients_server_socket, commaddress->ai_addr, commaddress->ai_addrlen)) { perror("bind 2"); return 23; } listen(clients_server_socket, 1); } if (!servermode) { if (!commlisten) { send_simple(MT_HELLO); } else { send_simple(MT_NOOP); } } { struct sockaddr sa; socklen_t sa_len = address->ai_addrlen; getsockname(s, &sa, &sa_len); char hbuf[NI_MAXHOST], sbuf[NI_MAXSERV]; gai_error = getnameinfo(&sa, sa_len, hbuf, sizeof hbuf, sbuf, sizeof sbuf, NI_DGRAM|NI_NUMERICHOST|NI_NUMERICSERV); if (!gai_error) { fprintf(stderr,"%s\n%s\n",hbuf,sbuf); fflush(stderr); } } int enable_udp_sending; /* Warning: Reusing "address" for peer's address */ enable_udp_sending = 1; if (servermode) { enable_udp_sending = 0; /* Wait for arrival of some peer's packet. We'll overwrite "address" with recvfrom's value and start sending things there */ } int period = 30; if (getenv("UDPMSTR_PERIOD")) period = atoi(getenv("UDPMSTR_PERIOD")); tcp_helper_period=5; if (getenv("UDPMSTR_TCPHELPER_PERIOD")) tcp_helper_period = atoi(getenv("UDPMSTR_TCPHELPER_PERIOD")); for(;;) { fd_set rfds; fd_set wfds; FD_ZERO(&rfds); FD_ZERO(&wfds); int maxfd = 0; #define add_to_rfds(x) \ if (x != -1) { \ FD_SET(x, &rfds); \ if (x >= maxfd) maxfd = x+1; \ } int currently_allowed_window_size = MAX_PACKET; if (tcp_helper_active) currently_allowed_window_size = MAX_UNACK; if (enable_udp_sending && unacknowledged_data < currently_allowed_window_size && !(tcp_helper_active && tcp_helper_send_debt_len)) { add_to_rfds(readfd) } add_to_rfds(s) add_to_rfds(clients_server_socket) add_to_rfds(tcp_helper_server_socket) if (!tcp_helper_packet_ready) { add_to_rfds(tcp_helper_socket) } if (tcp_helper_send_debt_len && tcp_helper_socket!=-1) { FD_SET(tcp_helper_socket, &wfds); } struct timeval *timeout = NULL; struct timeval tv; if (!servermode) { timeout = &tv; if (short_timeout) { tv.tv_sec = 0; tv.tv_usec = 100000; } else { tv.tv_sec = period; tv.tv_usec = 0; } } int ret = select(maxfd, &rfds, &wfds, NULL, timeout); if (ret==-1) { if (errno==EAGAIN || errno==EINTR) continue; perror("select"); return 91; } if (!ret) { /* timeout */ if (tcp_helper_active && !short_timeout) { fprintf(stderr, "Timing out TCP helper\n"); close_tcp_helper(); } if (unacknowledged_data>0) { send_data(SEND_FROM_BEGINNING); } if (short_timeout==5) { if (tcp_helper_active) { fprintf(stderr, "Probably problems with TCP helper\n"); close_tcp_helper(); } short_timeout=0; } if (short_timeout) { if (skipped_acks) { send_ack(); skipped_acks=0; } ++short_timeout; } send_simple(MT_REQUEST); } if (try_tcp_helper) { try_tcp_helper = 0; if (!tcp_helper_active) { open_tcp_helper(); } } if (tcp_helper_server_socket!=-1 && FD_ISSET(tcp_helper_server_socket, &rfds)) { close_tcp_helper(); struct sockaddr sa; socklen_t sa_len = sizeof(sa); tcp_helper_socket = accept(tcp_helper_server_socket, &sa, &sa_len); if (tcp_helper_socket!=-1) { nonblocking_socket(tcp_helper_socket); tcp_helper_active=1; send_simple(MT_NOOP); } } if (tcp_helper_socket != -1 && FD_ISSET(tcp_helper_socket, &wfds)) { if (tcp_helper_send_debt_len) { int ret = send(tcp_helper_socket, tcp_helper_send_debt, tcp_helper_send_debt_len, 0); if(ret==-1 && (errno==EINTR || errno==EAGAIN)) continue; if (ret<1) { fprintf(stderr, "TCP helper disconnected\n"); close_tcp_helper(); send_simple(MT_REQUEST); } memmove(tcp_helper_send_debt, tcp_helper_send_debt+ret, tcp_helper_send_debt_len-ret); tcp_helper_send_debt_len -= ret; } } if (!tcp_helper_packet_ready && tcp_helper_socket != -1 && FD_ISSET(tcp_helper_socket, &rfds)) { int ret; if (tcp_helper_pending_packet_length == -1) { if (tcp_helper_pending_packet_length_first_byte == -1) { unsigned char b; ret = recv(tcp_helper_socket, &b, 1, 0); if (ret==-1 && (errno==EINTR || errno==EAGAIN)) continue; if (ret!=1) { fprintf(stderr, "Short read or failure with TCP helper (ret=%d)\n", ret); close_tcp_helper(); send_simple(MT_REQUEST); continue; } tcp_helper_pending_packet_length_first_byte=b; } unsigned char b; ret = recv(tcp_helper_socket, &b, 1, 0); if (ret==-1 && (errno==EINTR || errno==EAGAIN)) continue; if (ret!=1) { fprintf(stderr, "Short read or failure with TCP helper (ret=%d)\n", ret); close_tcp_helper(); send_simple(MT_REQUEST); continue; } tcp_helper_pending_packet_length = b + 0x100 * tcp_helper_pending_packet_length_first_byte; tcp_helper_pending_packet_length_first_byte = -1; if (tcp_helper_pending_packet_length>2*MAX_PACKET) { fprintf(stderr, "Excess packet length on TCP helper\n"); close_tcp_helper(); send_simple(MT_REQUEST); continue; } } else { ret = recv(tcp_helper_socket, tcp_helper_debt_buf+tcp_helper_debt_buf_offset, tcp_helper_pending_packet_length-tcp_helper_debt_buf_offset, 0); if (ret==-1 && (errno==EINTR || errno==EAGAIN)) continue; if (ret<=0) { fprintf(stderr, "EOF on TCP helper\n"); close_tcp_helper(); send_simple(MT_REQUEST); continue; } tcp_helper_debt_buf_offset+=ret; if (tcp_helper_debt_buf_offset==tcp_helper_pending_packet_length) { //fprintf(stderr, "Ready a packet from debt: %d\n", tcp_helper_pending_packet_length); tcp_helper_packet_ready = 1; if (!tcp_helper_active) { fprintf(stderr, "Connected TCP helper\n"); } tcp_helper_active = 1; } } } if (clients_server_socket!=-1 && FD_ISSET(clients_server_socket, &rfds)) { /* listen and accept socket before continuing */ int s3 = accept(clients_server_socket, commaddress->ai_addr, &commaddress->ai_addrlen); if (s3 == -1) { perror("accept"); close(clients_server_socket); clients_server_socket = -1; return 23; } close(readfd); readfd = s3; writefd = s3; stale_mode_message_already_printed=0; seq = 0; ack_seq = 0; recv_seq = 0; startp = 0; endp = 0; unacknowledged_data = 0; memset(circular_buffer, 0, sizeof circular_buffer); server_received_hello = 1; send_simple(MT_HELLO); } if (readfd!= -1 && FD_ISSET(readfd, &rfds)) { int max_read_len = MAX_UNACK - unacknowledged_data; if (max_read_len > MAX_PACKET) max_read_len = MAX_PACKET; if (socket_ops) ret = recv(readfd, circular_buffer+endp, max_read_len, 0); else ret = read(readfd, circular_buffer+endp, max_read_len); if (ret==-1 && (errno==EINTR || errno==EAGAIN)) { // Ignore } else if (ret == 0 || ret == -1) { send_simple(MT_QUIT); if (!commmode) { return 0; } else { close(readfd); readfd = -1; writefd = -1; stale_mode_message_already_printed=0; } } else { endp+=ret; unacknowledged_data+=ret; seq+=ret; if(seq >= MAXSEQ) seq -= MAXSEQ; send_data(SEND_FROM_END); } } if (FD_ISSET(s, &rfds) || tcp_helper_packet_ready) { int tcp_helper_packet_was_ready = 0; unsigned char buf[MAX_PACKET * 2]; if (FD_ISSET(s, &rfds)) { ret = recvfrom(s, buf, sizeof buf, 0, address->ai_addr, &address->ai_addrlen); } else if (tcp_helper_packet_ready) { if (tcp_helper_pending_packet_length==-1) { fprintf(stderr, "Assertion failed: tcp_helper_packet_ready, but no tcp_helper_pending_packet_length\n"); close_tcp_helper(); send_simple(MT_REQUEST); continue; } ret = tcp_helper_pending_packet_length; memcpy(buf, tcp_helper_debt_buf, ret); tcp_helper_pending_packet_length = -1; tcp_helper_packet_ready = 0; tcp_helper_packet_was_ready = 1; tcp_helper_debt_buf_offset = 0; } if ((ret==-1 && (errno==EINTR || errno==EAGAIN)) || ret < 5) { // Ignore } else { unsigned long checksum = htonl (crc32buf(buf, ret-4)); if(memcmp(buf+ret-4, &checksum, 4)) { /* wrong checksum.*/ fprintf(stderr, "Wrong checksum for %c\n", buf[0]); if (buf[0] == MT_DATA) { send_simple(MT_REQUEST); } } else { ret-=4; enable_udp_sending = 1; switch(buf[0]) { case MT_QUIT: { if (commmode) { close(readfd); readfd = -1; writefd = -1; stale_mode_message_already_printed=0; continue; } else { return 0; } } break; case MT_ACK: { unsigned long l = ntohl(*(unsigned long *)(buf+1)); if (l>ack_seq || (ack_seq>=(MAXSEQ-MAX_UNACK) && (l <= MAX_UNACK))) { int confirmed_bytes; if (ack_seq>MAXSEQ ) { confirmed_bytes = l + MAXSEQ - ack_seq; ack_seq = l; } else { confirmed_bytes = l - ack_seq; ack_seq = l; } if (confirmed_bytes > MAX_UNACK || confirmed_bytes<0) { fprintf(stderr, "Invalid confirmed_bytes %d\n", confirmed_bytes); fprintf(stderr, "l=%lu seq=%lu ack_seq=%lu\n", l, seq, ack_seq); continue; } unacknowledged_data -= confirmed_bytes; startp += confirmed_bytes; if (startp<0 || endp < 0 || startp >= sizeof circular_buffer || endp >= sizeof circular_buffer) { fprintf(stderr, "Something wrong, exiting\n"); return 43; } if (startp>=MAX_UNACK/2) { memmove(circular_buffer, circular_buffer+startp, endp - startp); endp = endp - startp; startp = 0; } //fprintf(stderr, "ACK unacknowledged_data=%d cb=%d w=%lu\n", unacknowledged_data, confirmed_bytes, seq-ack_seq); if (seq!=ack_seq && !tcp_helper_active) { send_data(SEND_FROM_BEGINNING); } } } break; case MT_REQUEST: { if (!tcp_helper_packet_was_ready) { if (tcp_helper_active) fprintf(stderr, "Suspending TCP helper usage on MT_REQUEST\n"); tcp_helper_active=0; } send_data(SEND_FROM_BEGINNING); } break; case MT_DATA: { if (writefd == -1 || (servermode && !server_received_hello)) { if (!stale_mode_message_already_printed) { fprintf(stderr, "Looks like some stale session\n"); stale_mode_message_already_printed = 1; } send_simple(MT_QUIT); continue; } unsigned long l = ntohl(*(unsigned long *)(buf+1)); int unprocessed_len; if (recv_seq >= MAXSEQ - MAX_UNACK && l <= MAX_UNACK) { unprocessed_len = l + MAXSEQ - recv_seq; } else { unprocessed_len = l - recv_seq; } if (unprocessed_len < -MAX_UNACK || unprocessed_len > MAX_UNACK) { fprintf(stderr, "Invalid unprocessed_len %d\n", unprocessed_len); } if (unprocessed_len>0) { if (unprocessed_len > ret-5) { /* We lost some earlier packet */ send_simple(MT_REQUEST); /* Ask for re-sending */ fprintf(stderr, "This packet is not welcome now\n"); continue; /* Inhibit acknowledgement */ } else { safer_write(writefd, buf+ret-unprocessed_len, unprocessed_len); recv_seq = l; } if (!tcp_helper_packet_was_ready) { if (tcp_helper_active) fprintf(stderr, "Suspending TCP helper usage on MT_DATA\n"); tcp_helper_active=0; } } if (unprocessed_len == MAX_PACKET) { maybe_connect_tcp_helper(); } /* Send acknowledgement */ if (!tcp_helper_active) { send_ack(); } else { ++skipped_acks; if (skipped_acks >= MAX_UNACK/MAX_PACKET/3) { send_ack(); skipped_acks=0; } } if (unprocessed_len && !servermode) { short_timeout = 1; } else { short_timeout = 0; } } break; case MT_HELLO: { seq = 0; ack_seq = 0; recv_seq = 0; startp = 0; endp = 0; unacknowledged_data = 0; memset(circular_buffer, 0, sizeof circular_buffer); server_received_hello = 1; if (!tcp_helper_packet_was_ready) { if (tcp_helper_active) fprintf(stderr, "Suspending TCP helper usage on MT_HELLO\n"); tcp_helper_active=0; } if (commmode && !commlisten) { close(readfd); /* connect to the specified stream socket and use it */ int s2 = socket(commaddress->ai_family, commaddress->ai_socktype, commaddress->ai_protocol); if (connect(s2, commaddress->ai_addr, commaddress->ai_addrlen)) { perror("connect"); return 25; } readfd = s2; writefd = s2; stale_mode_message_already_printed=0; } else if (commmode) { fprintf(stderr, "HELLO received... Strange.\n"); } } break; case MT_NOOP: break; default: { fprintf(stderr, "Unknown message %c\n", buf[0]); } } // switch } } } } return 0; }