Socket-Client recv() gibt immer 0 zurück

Question

Ich versuche, einen HTTP-Proxy, wo nach dem GET/CONNET-Host-Namen in der HTTP-Anfrage einige Verbindungen höhere Prioritäten als andere haben.

Die Idee besteht darin, Anfragen mit höherer Priorität zu erfüllen, basierend auf einer gegebenen Liste von Hostnamen mit jeweils einer bestimmten Priorität.

ausstehende Verbindungen werden von accepter Gewinde in vier verschiedenen Warteschlangen (ein für jede Prioritätsklasse: maximale, mittlere, minimale und nicht klassifiziert) gespeichert werden; accepter wird dann fork() ein untergeordneter Prozess, der ausstehende Verbindungen in der Reihenfolge ihrer Priorität aus der Warteschlange herausnimmt und behandelt. Dadurch wird so, accepter Faden immer wieder neue Verbindungen annehmen und für jede Warteschlange gestellt conne Kurz gesagt, hier ist mein proxy:

• Haupt: öffnet TCP-Socket, bindet sich an einen bestimmten Port lauscht bis zu 10 Verbindungen, Anrufe Thread Accepter übergibt es den Socket fd geöffnet mit dem vorherigen socket() Aufruf und Joins für diesen Thread;
• accepter: dieser Thread bekommt die Buchse fd von Haupt geleitet und Schleifen mit accept() zurückkehr Client-Socket, recv() vom Client, analysiert die Anfrage und nach dem Hostnamen in der HTTP-Anforderung wird eine benutzerdefinierte struct von mir eine Warteschlange eingereiht werden, in der richtigen Warteschlange; Es wird dann fork() so ein Prozess wird aus der Warteschlange und die Verbindung zu behandeln;
• manageConnection: Dieser Prozess, gegabelt von accepter, entnimmt aus Warteschlangen prüft der knallte struct die Hostnamen Feld löst, öffnet ein Socket-Client, connets an den Server und, GET oder CONNECT, die Anforderung erfüllen.

New Proxy: nicht mehr fork() ich einen Thread-Pool von vier Threads (einem "accepter" und drei "connecter": da ich plane, diesen Proxy auf meiner RPi 2 zu setzen, die hat einen Quadcore-Prozessor, ich dachte, dass mindestens vier Threads gut waren). Ich habe jetzt eine mutex und zwei condition_variables. Der Code ist fast derselbe, außer für Threads, Mutexe und Bedingungsvariablen. Das sind neue Funktionen aufgerufen durch Gewinde:

• enqueue: dieser Thread enthält die accept() Schleife, wo es vom Client empfängt, analysiert die HTTP-Anforderung, findet den Hostnamen und entsprechend seiner Priorität, enqueue ein info_conn struct (typedefed am Anfang des Codes);

• dequeue: dieser Thread enthält die Warteschlangenauflösungs und Verwalten von Verbindungen Schleife, wo es eine info_conn struct aus einer Warteschlange erhält, ruft Client-Socket (die ich von accept() Schleife vor), löst Hostnamen und GET verwalten oder Verbindungsanforderung.

Das Problem: immer das gleiche, wenn es CONNECT Anfragen zu verwalten kommt, recv() von Client zurückgeben immer 0: Ich weiß, recv() 0 zurück, wenn die andere Seite der Verbindung getrennt hat, aber das ist nicht was ich wollte! Basierend auf einem Thread-Ansatz, ist dies ein triviales Producer/Consumer-Problem (herausspringen und in Warteschlangen pushen), daher denke ich, dass der Thread-Wechsel bei der Warteschlangen- und der Warteschlangenentfernung korrekt ist.

Mein (neu) Code:

#include <stdio.h> 
#include <string.h> 
#include <stdlib.h> 
#include <sys/socket.h> 
#include <sys/types.h> 
#include <sys/time.h> 
#include <arpa/inet.h> 
#include <unistd.h>  
#include <thread> 
#include <iostream> 
#include <netdb.h> 
#include <queue> 
#include <list> 
#include <vector> 
#include <condition_variable> 
#include <cstdlib> 


using namespace std; 

#define GET  0 
#define CONNECT 1 

#define DEFAULTCOLOR  "\033[0m" 
#define RED     "\033[22;31m" 
#define YELLOW    "\033[1;33m" 
#define GREEN    "\033[0;0;32m" 

#define MAX_SIZE   1000 
#define CONNECT_200_OK  "HTTP/1.1 200 Connection established\r\nProxy-agent: myproxy\r\n\r\n" 

// my custom struct stored in queues 
typedef struct info_connection { 
    int client_fd; 
    string host; 
    string payload; 
    int request; 
} info_conn; 

queue<info_conn>q1; 
queue<info_conn>q2; 
queue<info_conn>q3; 
queue<info_conn>q4; 
vector<thread> workers; 
condition_variable cond_read, cond_write; 
mutex mtx; 

void enqueue(int sock_client); 
void dequeue(void); 

int main(int argc, char *argv[]) { 
    int socket_desc; 
    struct sockaddr_in server; 

    socket_desc = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); 
    if (socket_desc == -1) { 
     perror("socket()"); 
     exit(-1); 
    } 

    server.sin_family = AF_INET; 
    server.sin_addr.s_addr = INADDR_ANY; 
    if (argc == 2) 
     server.sin_port = htons(atoi(argv[1])); 
    printf("listening to port %d\n", atoi(argv[1])); 

    if (bind(socket_desc,(struct sockaddr *)&server, sizeof(server)) < 0) { 
     perror("bind failed. Error"); 
     exit(-1); 
    } 
    printf("binded\n"); 

    listen(socket_desc, 10); 
    printf("listen\n"); 

    // thread pool, because I suck at forking 
    workers.push_back(thread(enqueue, socket_desc)); 
    workers.push_back(thread(dequeue)); 
    workers.push_back(thread(dequeue)); 
    workers.push_back(thread(dequeue)); 

    for (thread& t : workers) { 
     t.join(); 
    } 

    return 0; 
} 

void enqueue(int sock_client) { 
    printf("enqueue()\n"); 
    int client_sock; 
    struct sockaddr_in *client_struct; 
    unsigned int clilen; 
    bzero((char*)&client_struct, sizeof(client_struct)); 
    clilen = sizeof(client_struct); 
    char host_name[128]; 
    char buff[4096]; 
    int n_recv, n_send; 
    char *start_row, *end_row, *tmp_ptr, *tmp_start; 
    int req; 

    while((client_sock = accept(sock_client, (struct sockaddr *)&client_struct, &clilen))) { 
     memset(host_name, 0, sizeof(host_name)); 
     n_recv = recv(client_sock, buff, sizeof(buff), 0); 
     if (n_recv < 0) { 
      perror("recv()"); 
      break; 
     } 

     start_row = end_row = buff; 

     while ((end_row = strstr(start_row, "\r\n")) != NULL) { 
      int row_len = end_row - start_row; 
      if (row_len == 0) 
       break; 
      if (strncmp(buff, "GET ", 4) == 0) { 
       req = GET; 
       tmp_start = start_row + 4; 
       tmp_ptr = strstr(tmp_start, "//"); 
       int len = tmp_ptr - tmp_start; 
       tmp_start = tmp_start + len + 2; 
       tmp_ptr = strchr(tmp_start, '/'); 
       len = tmp_ptr - tmp_start; 
       strncpy(host_name, tmp_start, len); 
       break; 
      } 
      else if (strncmp(buff, "CONNECT ", 8) == 0) { 
       req = CONNECT; 
       tmp_start = start_row + 8; 
       tmp_ptr = strchr(tmp_start, ':'); 
       int host_len = tmp_ptr - tmp_start; 
       strncpy(host_name, tmp_start, host_len); 
       break; 
      } 
      start_row = end_row + 2; 
      /* if ((start_row - buff) >= strlen(buff)) 
        break;*/ 
     } 

     unique_lock<mutex> locker(mtx, defer_lock); 
     locker.lock(); 
     cond_write.wait(locker, [](){ 
      return (q1.size() < MAX_SIZE || q2.size() < MAX_SIZE || q3.size() < MAX_SIZE || q4.size() < MAX_SIZE); 
     }); 

     cout << "(DEBUG) thread " << this_thread::get_id() << " wants to insert, queues not full " << 
      q1.size() << ' ' << q2.size() << ' ' << q3.size() << ' ' << q4.size() << '\n'; 
     int priority = 0; 
     info_conn info_c; 
     info_c.client_fd = client_sock; 
     info_c.host = host_name; 
     info_c.request = req; 
     info_c.payload = string(buff); 
     cout << "(DEBUG) thread " << this_thread::get_id() << " looking for " << host_name << 
      " queues" << '\n'; 
     if (strcmp(host_name, "www.netflix.com") == 0) { 
      priority = 1; 
      printf("hostname = www.netflix.com, priority %d\n", priority); 
      q1.push(info_c); 
     } 
     else if (strcmp(host_name, "www.youtube.com") == 0) { 
      priority = 2; 
      printf("hostname = www.youtube.com, priority %d\n", priority); 
      q2.push(info_c); 
     } 
     else if (strcmp(host_name, "www.facebook.com") == 0) { 
      priority = 3; 
      printf("hostname = www.facebook.com, priority %d\n", priority); 
      q3.push(info_c); 
     } 
     else { 
      priority = 4; 
      printf("hostname %s not found in queues\n", host_name);     
      q4.push(info_c); 
     } 

     cout << GREEN << "(DEBUG) thread " << this_thread::get_id() << " inserted " << 
      q1.size() << ' ' << q2.size() << ' ' << q3.size() << ' ' << q4.size() << DEFAULTCOLOR<< '\n'; 

     locker.unlock(); 
     cond_read.notify_all(); 
    } 
    if (client_sock < 0) { 
     perror("accept failed"); 
     exit(-1); 
    } 
} 

void dequeue(void) { 
    int fd_client = -1; 
    int fd_server = -1; 
    struct sockaddr_in server; 
    int what_request; 
    char host_name[128]; 
    char buffer[1500]; 
    int n_send, n_recv; 
    size_t length; 
    info_conn req; 

    // CONNECT 
    int r, max; 
    int send_200_OK; 
    int read_from_client = 0; 
    int read_from_server = 0; 
    int send_to_client = 0; 
    int send_to_server = 0; 
    struct timeval timeout; 
    char buff[8192]; 
    fd_set fdset; 

    printf("dequeue()\n"); 
    while (true) { 
     unique_lock<mutex> locker(mtx, defer_lock); 
     locker.lock(); 
     cond_read.wait(locker, [](){ 
      return (q1.size() > 0 || q2.size() > 0 || q3.size() > 0 || q4.size() > 0); 
     }); 

     cout << "(DEBUG) thread " << this_thread::get_id() << " wants to remove, queues not empty " << 
      q1.size() << ' ' << q2.size() << ' ' << q3.size() << ' ' << q4.size() << '\n'; 
     if (q1.size() > 0) { 
      req = q1.front(); 
      q1.pop(); 
     } 
     else if (q2.size() > 0) { 
      req = q2.front(); 
      q2.pop(); 
     } 
     else if (q3.size() > 0) { 
      req = q3.front(); 
      q3.pop();  
     } 
     else if (q4.size() > 0) { 
      req = q4.front(); 
      q4.pop(); 
     } 
     cout << YELLOW <<"(DEBUG) thread " << this_thread::get_id() << " removed, " << 
      q1.size() << ' ' << q2.size() << ' ' << q3.size() << ' ' << q4.size() << DEFAULTCOLOR<<'\n'; 
     locker.unlock(); 
     // notify one, because I have only one "producer" thread 
     cond_write.notify_one(); 

     fd_client = req.client_fd; 
     //memcpy(host_name, req.host.c_str(), strlen(req.host)); 
     length = req.host.copy(host_name, req.host.size(), 0); 
     host_name[length] = '\0'; 
     what_request = req.request; 
     //memcpy(buffer, req.payload, req.payload.size()); 
     length = req.payload.copy(buffer, req.payload.size(), 0); 
     buffer[length] = '\0'; 
     what_request = req.request; 

     //cout << RED <<"(DEBUG) thread " << this_thread::get_id() << " copied packet payload " << 
     // buffer << DEFAULTCOLOR<<'\n'; 

     struct addrinfo* result; 
     struct addrinfo* res; 
     int error; 
     struct sockaddr_in *resolve; 

     fd_server = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); 
     if (fd_server < 0) { 
      perror("socket()"); 
      exit(-1); 
     } 

     cout << "(DEBUG) thread " << this_thread::get_id() << " fd_server " << fd_server << '\n'; 
     error = getaddrinfo(host_name, NULL, NULL, &result); 
     if (error != 0) { 
      if (error == EAI_SYSTEM) { 
       perror("getaddrinfo"); 
      } else { 
       fprintf(stderr, "error in getaddrinfo for (%s): %s\n", host_name, gai_strerror(error)); 
      } 
      exit(EXIT_FAILURE); 
     } 

     if (what_request == GET) { 
      server.sin_port = htons(80); 
     } 
     else if (what_request == CONNECT) { 
      server.sin_port = htons(443); 
     } 
     server.sin_family = AF_INET; 
     cout << "(DEBUG) thread " << this_thread::get_id() << " getaddrinfo()" << '\n'; 
     for (res = result; res != NULL; res = res->ai_next) { 
      if (res->ai_family == AF_INET) { 
       resolve = (struct sockaddr_in *)res->ai_addr; 
       //char *ip = inet_ntoa(resolve->sin_addr); 
       //printf("%s\n", ip); 
       server.sin_addr.s_addr = resolve->sin_addr.s_addr; 
       if (connect(fd_server, (struct sockaddr *)&server, sizeof (struct sockaddr_in)) < 0) { 
        fflush(stdout); 
        perror("connect()"); 
       } 
       else { 
        cout << "(DEBUG) thread " << this_thread::get_id() << " connected to " << inet_ntoa(server.sin_addr) << '\n'; 
       } 
       break; 
      } 
     } 

     // dealing with GET 
     if (what_request == GET) { 
      cout << "thread " << this_thread::get_id() << " dealing GET " << host_name << 
       " sending to server " << buffer << '\n'; 
      n_send = send(fd_server, buffer, strlen(buffer)+1, 0); 
      if (n_send < 0) { 
       cout << "thread " << this_thread::get_id() << " error sending GET request to server" << '\n'; 
       perror("send()"); 
       break; 
      } 
      do { 
       memset(buffer, 0, sizeof(buffer)); 
       n_recv = recv(fd_server, buffer, sizeof(buffer), 0); 
       cout << "thread " << this_thread::get_id() << " GET: " << host_name << " read from recv() " << n_recv << " bytes, " << 
        fd_client << "<->" << fd_server << '\n'; 
       n_send = send(fd_client, buffer, n_recv, 0); 
      } while (n_recv > 0); 

      if (n_recv < 0) { 
       cout << RED << "thread " << this_thread::get_id() << " error sending GET response from server to client" << DEFAULTCOLOR<<'\n'; 
       perror("send()"); 
       break; 
      } 
      close(fd_client); 
      close(fd_server); 
      cout << "thread " << this_thread::get_id() << 
       " done with GET request, quitting\n"; 
     } 

     // dealing with CONNECT 
     else if (what_request == CONNECT) { 
      cout << "thread " << this_thread::get_id() << " dealing CONNECT " << host_name << '\n';   
      max = fd_server >= fd_client ? fd_server+1 : fd_client+1; 
      send_200_OK = send(fd_client, CONNECT_200_OK, sizeof(CONNECT_200_OK), 0); 
      if (send_200_OK < 0) { 
       perror("send() 200 OK to client"); 
       break; 
      } 
      cout << "thread " << this_thread::get_id() << " SENT 200 OK to client " << '\n'; 

      int tot_recvd; 
      int tot_sent; 

      // TCP tunnel 
      while(true) { 
       memset(buff, 0, sizeof(buff)); 
       FD_ZERO(&fdset); 
       FD_SET(fd_client, &fdset); 
       FD_SET(fd_server, &fdset); 
       timeout.tv_sec = 15; 
       timeout.tv_usec = 0; 

       r = select(max, &fdset, NULL, NULL, &timeout); 

       if (r < 0) { 
        perror("select()"); 
        close(fd_client); 
        close(fd_server); 
        break; 
       } 

       if (r == 0) { // select timed out 
        printf("tunnel(): select() request timeout 408\n"); 
        close(fd_client); 
        close(fd_server); 
        break; 
       } 

       if (FD_ISSET(fd_client, &fdset)) { 
        tot_recvd = 0; 
        tot_sent = 0; 
        do { 
         read_from_client = recv(fd_client, &(buff[tot_recvd]), sizeof(buff), 0); 
         tot_recvd += read_from_client; 
         cout << "thread " << this_thread::get_id() << 
          " select(), reading from client " << fd_client << 
          " " << read_from_client << " bytes, " << fd_client<< " <-> " << fd_server<<'\n'; 
         if (buff[tot_recvd-1] == '\0') { 
          break; 
         } 
        } while (read_from_client > 0); 

        if (read_from_client < 0) { 
         perror("recv()"); 
         close(fd_client); 
         close(fd_server); 
         break; 
        } 

        if (read_from_client == 0) { 
         // this always happens!!! 
        } 

        send_to_server = send(fd_server, buff, read_from_client, 0); 
        if (send_to_server < 0) { 
         perror("send() to client"); 
         close(fd_client); 
         close(fd_server); 
         break; 
        } 
       } 

       if (FD_ISSET(fd_server, &fdset)) { 
        tot_recvd = 0; 
        tot_sent = 0; 
        do { 
         read_from_server = recv(fd_server, &(buff[tot_recvd]), sizeof(buff), 0); 
         tot_recvd += read_from_server; 
         cout << "thread " << this_thread::get_id() << 
          " select(), reading from server " << fd_client << 
          " " << read_from_server << " bytes, " << fd_client<< " <-> " << fd_server<<'\n'; 
         if (buff[tot_recvd-1] == '\0') { 
          break; 
         } 
        } while (read_from_server > 0); 

        if (read_from_server < 0) { 
         perror("read()"); 
         close(fd_client); 
         close(fd_server); 
         break;   
        } 

        if (read_from_server == 0) { 
         cout << "thread " << this_thread::get_id() << " select(), server closed conn" << '\n'; 
         close(fd_client); 
         close(fd_server); 
         break;      
        } 

        send_to_client = send(fd_client, buff, read_from_server, 0);  
        if (send_to_client < 0) { 
         perror("send() to client"); 
         close(fd_client); 
         close(fd_server); 
         break; 
        }  
       } 
      } 
      cout << "thread " << this_thread::get_id() << " done with CONNECT request\n"; 
     } 
    } 
} 

Umwelt: Proxy läuft auf meinem Laptop mit Ubuntu 14.04, x86_64; Der Proxy wurde in Chrome mit dem SwitchyOmega-Plugin getestet, mit dem der Datenverkehr auf einem bestimmten Port (derselbe Port, den ich an meinen Proxy weiterleiten kann) umgeleitet wird, kompiliert mit g++ -std=c++11 -pedantic -Wall -o funwithproxyfork funwithproxyfork.cpp -lpthread.

Output (versucht, für Netflix und YouTube, sie beide das gleiche Problem hat, das heißt client closed conn, recv() liefert 0):

req: 1, hostname: www.netflix.com, priority: 1 
thread 5611 accepting again 
(CHILD 5627) is about to handle conn 
(CHILD 5627) popped sock_client 4, request 1 
req: 1, hostname: www2-ext-s.nflximg.net, priority: 4 
thread 5611 accepting again 
(CHILD 5628) is about to handle conn 
(CHILD 5628) popped sock_client 4, request 1 
req: 1, hostname: www2-ext-s.nflximg.net, priority: 4 
thread 5611 accepting again 
(CHILD 5629) is about to handle conn 
(CHILD 5629) popped sock_client 4, request 1 
(CHILD 5627) attempting to connect to 54.247.92.196 (www.netflix.com) 
(CHILD 5628) attempting to connect to 54.247.125.40 (www.netflix.com) 
(CHILD 5629) attempting to connect to 54.247.110.247 (www.netflix.com) 
(CHILD 5627) connected to www.netflix.com, dealing CONNECT request 
(CHILD 5628) connected to www.netflix.com, dealing CONNECT request 
(CHILD 5628) client closed conn 
(CHILD 5627) client closed conn 
(CHILD 5628) done with CONNECT request 
(CHILD 5627) done with CONNECT request 
req: 1, hostname: www.netflix.com, priority: 1 
thread 5611 accepting again 
(CHILD 5630) is about to handle conn 
(CHILD 5630) popped sock_client 4, request 1 
(CHILD 5630) attempting to connect to 176.34.188.125 (www.netflix.com) 
(CHILD 5629) connected to www.netflix.com, dealing CONNECT request 
(CHILD 5629) client closed conn 
(CHILD 5629) done with CONNECT request 
(CHILD 5630) connected to www.netflix.com, dealing CONNECT request 

Dann sagt es nichts anderes.

Answer 1

Aus der Untersuchung des Codes scheint dies normal zu sein und aufgrund der Art, wie HTTP/1.1 funktioniert.

Sie verwenden wahrscheinlich einige Clients, die HTTP/1.1-Pipelining unterstützen. Wenn HTTP/1.1-Pipelining aktiv ist, hält der Server die Verbindung offen, falls der Client eine weitere Anfrage senden möchte. Wenn der Client dies nicht tut, schließt der Client die Verbindung.

Offenbar erwartet Ihr Code, dass der Server die Verbindung beendet, nachdem er auf die HTTP-Anforderung geantwortet hat, und Sie erwarten nicht, dass der Client seine Seite der Verbindung zuerst schließt. Dies gilt nicht für HTTP/1.1, bei dem entweder der Client oder der Server die Verbindung zuerst schließen kann. Entweder ist das Schließen der Verbindung normal.

Also, es gibt kein Problem hier, außer für einige Probleme, die ich separat in den Kommentaren notiert habe, ohne Bezug auf die recv() Problem zur Hand. Darüber hinaus überprüft der Code an vielen Stellen nicht ausreichend den Rückgabewert von send(), und es nimmt an, dass alle angeforderten Byte gesendet worden sind. Das ist falsch. Sie können nicht garantieren, dass send() die genaue Anzahl der angeforderten Bytes sendet. Es kann tatsächlich weniger senden, und dies wird im Rückgabewert angegeben, der die Anzahl der tatsächlich gesendeten Bytes angibt.

Dieser Proxy wird bei hoher Datenverkehrslast fehlschlagen, da weniger Bytes gesendet werden, als angefordert wurde. Der Code kann diese Situation jedoch nicht erkennen und kann nicht korrekt verarbeitet werden. Es wird zum Beispiel 2000 Bytes vom Server lesen, versuchen, sie an den Client zu senden send() berichtet, dass 1000 Bytes gesendet wurden, der Code geht auf seinem fröhlichen Weg weiter, und der Client erhält nicht die gesamte Antwort vom Server. Hillarität folgt.

Außerdem gibt es hier einige andere Race-Bedingungen, die dazu führen können, dass der Proxy mit HTTP/1.1-Clients, die das Pipelining vollständig unterstützen, "verkeilt" oder gesperrt wird. Aber wenn Sie diese Art von Problemen haben, wird dies eine andere Frage sein müssen ...