Bugfixes and enhancements in sendmes and dns farm

svn:r161

7 files changed, 88 insertions, 75 deletions

diff --git a/src/or/circuit.c b/src/or/circuit.c
index bdb168820..fc3ec1ac2 100644
--- a/src/or/circuit.c
+++ b/src/or/circuit.c
@@ -279,17 +279,18 @@ int circuit_deliver_data_cell_from_edge(cell_t *cell, circuit_t *circ, char edge
     numsent_ap++;
     log(LOG_DEBUG,"circuit_deliver_data_cell_from_edge(): now sent %d data cells from ap", numsent_ap);
     if(circ->p_receive_circwindow <= 0) {
-      log(LOG_DEBUG,"circuit_deliver_data_cell_from_edge(): window 0, queueing for later.");
+      log(LOG_DEBUG,"circuit_deliver_data_cell_from_edge(): pwindow 0, queueing for later.");
       circ->data_queue = data_queue_add(circ->data_queue, cell);
       return 0;
     }
     circ->p_receive_circwindow--;
+//    log(LOG_INFO,"circuit_deliver_data_cell_from_edge(): p_receive_circwindow now %d.",circ->p_receive_circwindow);
   } else { /* i'm the exit */
     cell_direction = CELL_DIRECTION_IN;
     numsent_exit++;
     log(LOG_DEBUG,"circuit_deliver_data_cell_from_edge(): now sent %d data cells from exit", numsent_exit);
     if(circ->n_receive_circwindow <= 0) {
-      log(LOG_DEBUG,"circuit_deliver_data_cell_from_edge(): window 0, queueing for later.");
+      log(LOG_DEBUG,"circuit_deliver_data_cell_from_edge(): nwindow 0, queueing for later.");
       circ->data_queue = data_queue_add(circ->data_queue, cell);
       return 0;
     }
@@ -299,7 +300,7 @@ int circuit_deliver_data_cell_from_edge(cell_t *cell, circuit_t *circ, char edge
   if(circuit_deliver_data_cell(cell, circ, cell_direction) < 0) {
     return -1;
   }
-    
+
   circuit_consider_stop_edge_reading(circ, edge_type); /* has window reached 0? */
   return 0;
 }
@@ -420,18 +421,13 @@ void circuit_resume_edge_reading(circuit_t *circ, int edge_type) {
 
   assert(edge_type == EDGE_EXIT || edge_type == EDGE_AP);
 
-  if(edge_type == EDGE_EXIT)
-    conn = circ->n_conn;
-  else
-    conn = circ->p_conn;
-
   /* first, send the queue waiting at circ onto the circuit */
   while(circ->data_queue) {
     assert(circ->data_queue->cell);
-    if(edge_type == EDGE_EXIT) {   
-      circ->p_receive_circwindow--;
-      assert(circ->p_receive_circwindow >= 0);
-      
+    if(edge_type == EDGE_EXIT) {
+      circ->n_receive_circwindow--;
+      assert(circ->n_receive_circwindow >= 0);
+
       if(circuit_deliver_data_cell(circ->data_queue->cell, circ, CELL_DIRECTION_IN) < 0) {
         circuit_close(circ);
         return;
@@ -439,8 +435,8 @@ void circuit_resume_edge_reading(circuit_t *circ, int edge_type) {
     } else { /* ap */
       circ->p_receive_circwindow--;
       assert(circ->p_receive_circwindow >= 0);
-      
-      if(circuit_deliver_data_cell(circ->data_queue->cell, circ, CELL_DIRECTION_IN) < 0) {
+
+      if(circuit_deliver_data_cell(circ->data_queue->cell, circ, CELL_DIRECTION_OUT) < 0) {
         circuit_close(circ);
         return;
       }
@@ -454,6 +450,11 @@ void circuit_resume_edge_reading(circuit_t *circ, int edge_type) {
       return;
   }
 
+  if(edge_type == EDGE_EXIT)
+    conn = circ->n_conn;
+  else
+    conn = circ->p_conn;
+
   for( ; conn; conn=conn->next_topic) {
     if((edge_type == EDGE_EXIT && conn->n_receive_topicwindow > 0) ||
        (edge_type == EDGE_AP   && conn->p_receive_topicwindow > 0)) {
@@ -470,10 +471,10 @@ int circuit_consider_stop_edge_reading(circuit_t *circ, int edge_type) {
 
   assert(edge_type == EDGE_EXIT || edge_type == EDGE_AP);
 
-  if(edge_type == EDGE_EXIT && circ->p_receive_circwindow <= 0)
-    conn = circ->n_conn;      
-  else if(edge_type == EDGE_AP && circ->n_receive_circwindow <= 0)
-    conn = circ->p_conn;      
+  if(edge_type == EDGE_EXIT && circ->n_receive_circwindow <= 0)
+    conn = circ->n_conn;
+  else if(edge_type == EDGE_AP && circ->p_receive_circwindow <= 0)
+    conn = circ->p_conn;
   else
     return 0;
 
@@ -492,18 +493,22 @@ int circuit_consider_sending_sendme(circuit_t *circ, int edge_type) {
   sendme.length = CIRCWINDOW_INCREMENT;
 
   if(edge_type == EDGE_AP) { /* i'm the AP */
-    if(circ->n_receive_circwindow < CIRCWINDOW_START-CIRCWINDOW_INCREMENT) {
-      log(LOG_DEBUG,"circuit_consider_sending_sendme(): Queueing sendme forward.");
+    while(circ->n_receive_circwindow < CIRCWINDOW_START-CIRCWINDOW_INCREMENT) {
+      log(LOG_INFO,"circuit_consider_sending_sendme(): n_receive_circwindow %d, Queueing sendme forward.", circ->n_receive_circwindow);
       circ->n_receive_circwindow += CIRCWINDOW_INCREMENT;
       sendme.aci = circ->n_aci;
-      return connection_write_cell_to_buf(&sendme, circ->n_conn); /* (clobbers sendme) */
+      if(connection_write_cell_to_buf(&sendme, circ->n_conn) < 0) {
+        return -1;
+      }
     }
   } else if(edge_type == EDGE_EXIT) { /* i'm the exit */
-    if(circ->p_receive_circwindow < CIRCWINDOW_START-CIRCWINDOW_INCREMENT) {
-      log(LOG_DEBUG,"circuit_consider_sending_sendme(): Queueing sendme back.");
+    while(circ->p_receive_circwindow < CIRCWINDOW_START-CIRCWINDOW_INCREMENT) {
+      log(LOG_INFO,"circuit_consider_sending_sendme(): p_receive_circwindow %d, Queueing sendme back.", circ->p_receive_circwindow);
       circ->p_receive_circwindow += CIRCWINDOW_INCREMENT;
       sendme.aci = circ->p_aci;
-      return connection_write_cell_to_buf(&sendme, circ->p_conn); /* (clobbers sendme) */
+      if(connection_write_cell_to_buf(&sendme, circ->p_conn) < 0) {
+        return -1;
+      }
     }
   }
   return 0;
@@ -548,7 +553,7 @@ void circuit_about_to_close_connection(connection_t *conn) {
     cell.length = TOPIC_HEADER_SIZE;
     *(uint16_t *)(cell.payload+2) = htons(conn->topic_id);
     *cell.payload = TOPIC_COMMAND_END;
-   
+
     if(conn == circ->p_conn) {
       circ->p_conn = conn->next_topic;
       edge_type = EDGE_AP;
diff --git a/src/or/command.c b/src/or/command.c
index 944a4f0ad..7d2b1b939 100644
--- a/src/or/command.c
+++ b/src/or/command.c
@@ -134,8 +134,8 @@ void command_process_create_cell(cell_t *cell, connection_t *conn) {
     }
     memcpy((void *)(circ->onion+circ->recvlen),(void *)cell->payload,cell->length);
     circ->recvlen += cell->length;
-    log(LOG_DEBUG,"command_process_create_cell(): Secondary create cell handled, have received %d of %d onion bytes.",
-        circ->recvlen,circ->onionlen);
+    log(LOG_DEBUG,"command_process_create_cell(): Secondary create cell handled, have received %d of %d onion bytes (aci %d)",
+        circ->recvlen,circ->onionlen,circ->p_aci);
   }
 
   if(circ->recvlen != circ->onionlen) {
@@ -181,25 +181,26 @@ void command_process_sendme_cell(cell_t *cell, connection_t *conn) {
   if(cell->aci == circ->p_aci) { /* it's an outgoing cell */
     circ->n_receive_circwindow += cell->length;
     assert(circ->n_receive_circwindow <= CIRCWINDOW_START);
-    log(LOG_DEBUG,"connection_process_sendme_cell(): n_receive_circwindow for aci %d is %d.",circ->n_aci,circ->n_receive_circwindow);
+    log(LOG_INFO,"command_process_sendme_cell(): n_receive_circwindow for aci %d is %d.",circ->n_aci,circ->n_receive_circwindow);
     if(!circ->n_conn || circ->n_conn->type == CONN_TYPE_EXIT) {
       circuit_resume_edge_reading(circ, EDGE_EXIT);
     } else {
       cell->aci = circ->n_aci; /* switch it */
-      if(connection_write_cell_to_buf(cell, circ->n_conn) < 0) { /* (clobbers cell) */
+      if(connection_write_cell_to_buf(cell, circ->n_conn) < 0) {
         circuit_close(circ);
         return;
       }
     }
   } else { /* it's an ingoing cell */
+    assert(cell->aci == circ->n_aci);
     circ->p_receive_circwindow += cell->length;
-    log(LOG_DEBUG,"connection_process_sendme_cell(): p_receive_circwindow for aci %d is %d.",circ->p_aci,circ->p_receive_circwindow);
+    log(LOG_INFO,"command_process_sendme_cell(): p_receive_circwindow for aci %d is %d.",circ->p_aci,circ->p_receive_circwindow);
     assert(circ->p_receive_circwindow <= CIRCWINDOW_START);
     if(!circ->p_conn || circ->p_conn->type == CONN_TYPE_AP) {
       circuit_resume_edge_reading(circ, EDGE_AP);
     } else {
       cell->aci = circ->p_aci; /* switch it */
-      if(connection_write_cell_to_buf(cell, circ->p_conn) < 0) { /* (clobbers cell) */
+      if(connection_write_cell_to_buf(cell, circ->p_conn) < 0) {
         circuit_close(circ);
         return;
       }
@@ -217,30 +218,42 @@ void command_process_data_cell(cell_t *cell, connection_t *conn) {
     return;
   }
 
-  if(circ->state == CIRCUIT_STATE_ONION_WAIT) {
-    log(LOG_DEBUG,"command_process_data_cell(): circuit in onion_wait. Dropping data cell.");
-    return;
-  }
-  if(circ->state == CIRCUIT_STATE_OR_WAIT) {
-    log(LOG_DEBUG,"command_process_data_cell(): circuit in or_wait. Dropping data cell.");
-    return;
-  }
   if(circ->state == CIRCUIT_STATE_ONION_PENDING) {
     log(LOG_DEBUG,"command_process_data_cell(): circuit in create_wait. Queueing data cell.");
     onion_pending_data_add(circ, cell);
     return;
   }
 
-  /* circ->p_conn and n_conn are only null if we're at an edge point with no connections yet */
-
   if(cell->aci == circ->p_aci) { /* it's an outgoing cell */
-    cell->aci = circ->n_aci; /* switch it */
     if(--circ->p_receive_circwindow < 0) { /* is it less than 0 after decrement? */
-      log(LOG_INFO,"connection_process_data_cell(): Too many data cells for circuit (aci %d). Closing.", circ->p_aci);
+      log(LOG_INFO,"connection_process_data_cell(): Too many data cells for out circuit (aci %d). Closing.", circ->p_aci);
       circuit_close(circ);
       return;
     }
     log(LOG_DEBUG,"connection_process_data_cell(): p_receive_circwindow for aci %d is %d.",circ->p_aci,circ->p_receive_circwindow);
+  }
+
+  if(cell->aci == circ->n_aci) { /* it's an ingoing cell */
+    if(--circ->n_receive_circwindow < 0) { /* is it less than 0 after decrement? */
+      log(LOG_INFO,"connection_process_data_cell(): Too many data cells for in circuit (aci %d). Closing.", circ->n_aci);
+      circuit_close(circ);
+      return;
+    }
+    log(LOG_DEBUG,"connection_process_data_cell(): n_receive_circwindow for aci %d is %d.",circ->n_aci,circ->n_receive_circwindow);
+  }
+
+  if(circ->state == CIRCUIT_STATE_ONION_WAIT) {
+    log(LOG_WARNING,"command_process_data_cell(): circuit in onion_wait. Dropping data cell.");
+    return;
+  }
+  if(circ->state == CIRCUIT_STATE_OR_WAIT) {
+    log(LOG_WARNING,"command_process_data_cell(): circuit in or_wait. Dropping data cell.");
+    return;
+  }
+  /* circ->p_conn and n_conn are only null if we're at an edge point with no connections yet */
+
+  if(cell->aci == circ->p_aci) { /* it's an outgoing cell */
+    cell->aci = circ->n_aci; /* switch it */
     if(circuit_deliver_data_cell(cell, circ, CELL_DIRECTION_OUT) < 0) {
       log(LOG_INFO,"command_process_data_cell(): circuit_deliver_data_cell (forward) failed. Closing.");
       circuit_close(circ);
@@ -248,14 +261,8 @@ void command_process_data_cell(cell_t *cell, connection_t *conn) {
     }
   } else { /* it's an ingoing cell */
     cell->aci = circ->p_aci; /* switch it */
-    if(--circ->n_receive_circwindow < 0) { /* is it less than 0 after decrement? */
-      log(LOG_DEBUG,"connection_process_data_cell(): Too many data cells for circuit (aci %d). Closing.", circ->n_aci);
-      circuit_close(circ);
-      return;
-    }
-    log(LOG_DEBUG,"connection_process_data_cell(): n_receive_circwindow for aci %d is %d.",circ->n_aci,circ->n_receive_circwindow);
     if(circuit_deliver_data_cell(cell, circ, CELL_DIRECTION_IN) < 0) {
-      log(LOG_DEBUG,"command_process_data_cell(): circuit_deliver_data_cell (backward to AP) failed. Closing.");
+      log(LOG_DEBUG,"command_process_data_cell(): circuit_deliver_data_cell (backward) failed. Closing.");
       circuit_close(circ);
       return;
     }
diff --git a/src/or/connection.c b/src/or/connection.c
index fd16c195c..0e7847a64 100644
--- a/src/or/connection.c
+++ b/src/or/connection.c
@@ -512,7 +512,7 @@ int connection_send_destroy(aci_t aci, connection_t *conn) {
   return connection_write_cell_to_buf(&cell, conn);
 }
 
-int connection_write_cell_to_buf(cell_t *cellp, connection_t *conn) {
+int connection_write_cell_to_buf(const cell_t *cellp, connection_t *conn) {
   char networkcell[CELL_NETWORK_SIZE];
   char *n = networkcell;
  
diff --git a/src/or/connection_ap.c b/src/or/connection_ap.c
index 2173d0998..43bf960c6 100644
--- a/src/or/connection_ap.c
+++ b/src/or/connection_ap.c
@@ -128,6 +128,7 @@ int ap_handshake_process_socks(connection_t *conn) {
   /* now we're all ready to make an onion or send a begin */
 
   if(circ && circ->state == CIRCUIT_STATE_OPEN) {
+    /* FIXME if circ not yet open, figure out how to queue this begin? */
     /* add it into the linked list of topics on this circuit */
     log(LOG_DEBUG,"ap_handshake_process_socks(): attaching new conn to circ. n_aci %d.", circ->n_aci);
     conn->next_topic = circ->p_conn;
@@ -308,12 +309,12 @@ int ap_handshake_send_onion(connection_t *ap_conn, connection_t *n_conn, circuit
     if(dataleft >= CELL_PAYLOAD_SIZE) {
       cell.length = CELL_PAYLOAD_SIZE;
       memcpy(cell.payload, tmpbuf + tmpbuflen - dataleft, CELL_PAYLOAD_SIZE);
-      connection_write_cell_to_buf(&cell, n_conn); /* clobbers cell */
+      connection_write_cell_to_buf(&cell, n_conn);
       dataleft -= CELL_PAYLOAD_SIZE;
     } else { /* last cell */
       cell.length = dataleft;
       memcpy(cell.payload, tmpbuf + tmpbuflen - dataleft, dataleft);
-      connection_write_cell_to_buf(&cell, n_conn); /* clobbers cell */
+      connection_write_cell_to_buf(&cell, n_conn);
       dataleft = 0;
     }
   }
@@ -383,7 +384,7 @@ int connection_ap_process_data_cell(cell_t *cell, circuit_t *circ) {
   topic_id = ntohs(*(uint16_t *)(cell->payload+2));
   log(LOG_DEBUG,"connection_ap_process_data_cell(): command %d topic %d", topic_command, topic_id);
   num_seen++;
-  log(LOG_DEBUG,"connection_exit_process_data_cell(): Now seen %d data cells here.", num_seen);
+  log(LOG_DEBUG,"connection_ap_process_data_cell(): Now seen %d data cells here.", num_seen);
 
 
   circuit_consider_sending_sendme(circ, EDGE_AP);
@@ -452,7 +453,7 @@ int connection_ap_process_data_cell(cell_t *cell, circuit_t *circ) {
       break;
     case TOPIC_COMMAND_SENDME:
       if(!conn) {
-        log(LOG_DEBUG,"connection_exit_process_data_cell(): sendme cell dropped, unknown topic %d.",topic_id);
+        log(LOG_DEBUG,"connection_ap_process_data_cell(): sendme cell dropped, unknown topic %d.",topic_id);
         return 0;
       }
       conn->p_receive_topicwindow += TOPICWINDOW_INCREMENT;
diff --git a/src/or/connection_exit.c b/src/or/connection_exit.c
index f6fa27ec1..fb0daadc4 100644
--- a/src/or/connection_exit.c
+++ b/src/or/connection_exit.c
@@ -50,7 +50,7 @@ int connection_exit_finished_flushing(connection_t *conn) {
 
       log(LOG_DEBUG,"connection_exit_finished_flushing(): Connection to %s:%u established.",
           conn->address,conn->port);
-      
+
       conn->state = EXIT_CONN_STATE_OPEN;
       connection_watch_events(conn, POLLIN); /* stop writing, continue reading */
       if(connection_wants_to_flush(conn)) /* in case there are any queued data cells */
@@ -124,8 +124,8 @@ int connection_exit_begin_conn(cell_t *cell, circuit_t *circ) {
   log(LOG_DEBUG,"connection_exit_begin_conn(): Creating new exit connection.");
   n_conn = connection_new(CONN_TYPE_EXIT);
   if(!n_conn) {
-    log(LOG_DEBUG,"connection_exit_begin_conn(): connection_new failed. Closing.");
-    return -1;
+    log(LOG_DEBUG,"connection_exit_begin_conn(): connection_new failed. Dropping.");
+    return 0;
   }
 
   cell->payload[0] = 0;
@@ -140,9 +140,9 @@ int connection_exit_begin_conn(cell_t *cell, circuit_t *circ) {
   n_conn->n_receive_topicwindow = TOPICWINDOW_START;
   n_conn->p_receive_topicwindow = TOPICWINDOW_START;
   if(connection_add(n_conn) < 0) { /* no space, forget it */
-    log(LOG_DEBUG,"connection_exit_begin_conn(): connection_add failed. Closing.");
+    log(LOG_DEBUG,"connection_exit_begin_conn(): connection_add failed. Dropping.");
     connection_free(n_conn);
-    return -1;
+    return 0;
   }
 
   /* add it into the linked list of topics on this circuit */
diff --git a/src/or/dns.c b/src/or/dns.c
index d0f65e0ff..a728751a7 100644
--- a/src/or/dns.c
+++ b/src/or/dns.c
@@ -448,28 +448,24 @@ void init_cache_tree(void) {
  * dns farm, and return 0.
  */
 int dns_resolve(connection_t *exitconn) {
-  struct cached_resolve *new_resolve;
   struct cached_resolve *resolve;
+  struct cached_resolve search;
   struct pending_connection_t *pending_connection;
 
-  new_resolve = malloc(sizeof(struct cached_resolve));
-  memset(new_resolve, 0, sizeof(struct cached_resolve));
-  strncpy(new_resolve->question, exitconn->address, MAX_ADDRESSLEN);
+  strncpy(search.question, exitconn->address, MAX_ADDRESSLEN);
 
   /* try adding it to the tree. if it's already there it will
    * return it. */
-  resolve = SPLAY_INSERT(cache_tree, &cache_root, new_resolve);
+  resolve = SPLAY_FIND(cache_tree, &cache_root, &search);
   if(resolve) { /* already there. free up new_resolve */
-    free(new_resolve);
     switch(resolve->state) {
       case CACHE_STATE_PENDING:
         /* add us to the pending list */
         pending_connection = malloc(sizeof(struct pending_connection_t));
         pending_connection->conn = exitconn;
-        pending_connection->next = new_resolve->pending_connections;
-        new_resolve->pending_connections = pending_connection;
-
-        return dns_tor_to_master(exitconn);
+        pending_connection->next = resolve->pending_connections;
+        resolve->pending_connections = pending_connection;
+        return 0;
       case CACHE_STATE_VALID:
         exitconn->addr = resolve->answer;
         return connection_exit_connect(exitconn);
@@ -477,14 +473,18 @@ int dns_resolve(connection_t *exitconn) {
         return -1;
     }
   } else { /* this was newly added to the tree. ask the dns farm. */
-    new_resolve->state = CACHE_STATE_PENDING;
+    resolve = malloc(sizeof(struct cached_resolve));
+    memset(resolve, 0, sizeof(struct cached_resolve));
+    resolve->state = CACHE_STATE_PENDING;
+    strncpy(resolve->question, exitconn->address, MAX_ADDRESSLEN);
 
     /* add us to the pending list */
     pending_connection = malloc(sizeof(struct pending_connection_t));
     pending_connection->conn = exitconn;
-    pending_connection->next = new_resolve->pending_connections;
-    new_resolve->pending_connections = pending_connection;
+    pending_connection->next = resolve->pending_connections;
+    resolve->pending_connections = pending_connection;
 
+    SPLAY_INSERT(cache_tree, &cache_root, resolve);
     return dns_tor_to_master(exitconn);
   }
 
diff --git a/src/or/or.h b/src/or/or.h
index 2cd64e0c7..5fa776961 100644
--- a/src/or/or.h
+++ b/src/or/or.h
@@ -42,7 +42,7 @@
 #include "../common/ss.h"
 #include "../common/version.h"
 
-#define MAXCONNECTIONS 200 /* upper bound on max connections.
+#define MAXCONNECTIONS 1000 /* upper bound on max connections.
                               can be lowered by config file */
 
 #define MAX_BUF_SIZE (640*1024)
@@ -554,7 +554,7 @@ int connection_state_is_open(connection_t *conn);
 int connection_send_destroy(aci_t aci, connection_t *conn);
 int connection_send_connected(aci_t aci, connection_t *conn);
 int connection_encrypt_cell(char *cellp, connection_t *conn);
-int connection_write_cell_to_buf(cell_t *cellp, connection_t *conn);
+int connection_write_cell_to_buf(const cell_t *cellp, connection_t *conn);
 
 int connection_process_inbuf(connection_t *conn);
 int connection_package_raw_inbuf(connection_t *conn);