Fix, initialise read / write handles so we don't close fd=0

fdqueue.c

@@ -328,8 +328,10 @@ int fd_poll()
 	if (errno == ENXIO) fds[i].revents|=POLLERR;
 	call_alarm(fd_callbacks[i], fds[i].revents);
 	/* The alarm may have closed and unwatched the descriptor, make sure this descriptor still matches */
-	if (i<fdcount && fds[i].fd == fd)
-	  set_block(fds[i].fd);
+	if (i<fdcount && fds[i].fd == fd){
+	  if (set_block(fds[i].fd))
+	    FATALF("Alarm %p %s has a bad descriptor that wasn't closed!", fd_callbacks[i], alloca_alarm_name(fd_callbacks[i]));
+	}
       }
     }
   }

monitor.c

@@ -124,7 +124,8 @@ int monitor_setup_sockets()
   return 0;
   
   error:
-  close(sock);
+  if (sock>=0)
+    close(sock);
   return -1;
 }
 

overlay_interface.c

@@ -476,10 +476,8 @@ overlay_interface_init(const char *name, struct in_addr src_addr, struct in_addr
     
     strbuf d = strbuf_local(read_file, sizeof read_file);
     strbuf_path_join(d, serval_instancepath(), config.server.interface_path, ifconfig->file, NULL);
-    if (strbuf_overrun(d)){
-      WHYF("interface file name overrun: %s", alloca_str_toprint(strbuf_str(d)));
-      goto cleanup;
-    }
+    if (strbuf_overrun(d))
+      return WHYF("interface file name overrun: %s", alloca_str_toprint(strbuf_str(d)));
     
     if ((interface->alarm.poll.fd = open(read_file, O_APPEND|O_RDWR)) == -1) {
       if (errno == ENOENT && ifconfig->socket_type == SOCK_FILE) {
@@ -673,7 +671,7 @@ static void interface_read_file(struct overlay_interface *interface)
      but don't completely prevent other high priority alarms */
     if (interface->alarm.alarm == -1 || now < interface->alarm.alarm){
       interface->alarm.alarm = now;
-      interface->alarm.deadline = interface->alarm.alarm + 10000;
+      interface->alarm.deadline = interface->alarm.alarm + (new_packets > 10?50:10000);
     }
   }
   OUT();

rhizome_fetch.c

@@ -539,13 +539,16 @@ static int schedule_fetch(struct rhizome_fetch_slot *slot)
   int sock = -1;
   /* TODO Don't forget to implement resume */
   slot->start_time=gettime_ms();
+  slot->alarm.poll.fd = -1;
+  slot->write_state.blob_fd=-1;
+  slot->write_state.blob_rowid=-1;
+
   if (create_rhizome_import_dir() == -1)
     RETURN(WHY("Unable to create import directory"));
   if (slot->manifest) {
     if (rhizome_open_write(&slot->write_state, slot->manifest->fileHexHash, slot->manifest->fileLength, RHIZOME_PRIORITY_DEFAULT))
       RETURN(-1);
   } else {
-    slot->write_state.blob_rowid=-1;
     slot->write_state.file_offset=0;
     slot->write_state.file_length=-1;
   }
@@ -1343,7 +1346,7 @@ int rhizome_write_content(struct rhizome_fetch_slot *slot, char *buffer, int byt
     bytes=slot->write_state.file_length-(slot->write_state.file_offset+slot->write_state.data_size);
   }
 
-  if (slot->write_state.blob_rowid==-1) {
+  if (!slot->manifest){
     /* We are reading a manifest.  Read it into a buffer. */
     int count=bytes;
     if (count+slot->manifest_bytes>1024) count=1024-slot->manifest_bytes;

rhizome_http.c

@@ -261,7 +261,7 @@ void rhizome_server_poll(struct sched_ent *alarm)
     struct sockaddr addr;
     unsigned int addr_len = sizeof addr;
     int sock;
-    while ((sock = accept(rhizome_server_socket, &addr, &addr_len)) != -1) {
+    if ((sock = accept(rhizome_server_socket, &addr, &addr_len)) != -1) {
       struct sockaddr_in *peerip=NULL;
       if (addr.sa_family == AF_INET) {
 	peerip = (struct sockaddr_in *)&addr;
@@ -289,6 +289,8 @@ void rhizome_server_poll(struct sched_ent *alarm)
 	/* We are now trying to read the HTTP request */
 	request->request_type=RHIZOME_HTTP_REQUEST_RECEIVING;
 	request->alarm.function = rhizome_client_poll;
+	request->read_state.blob_fd=-1;
+	request->read_state.blob_rowid=-1;
 	connection_stats.name="rhizome_client_poll";
 	request->alarm.stats=&connection_stats;
 	request->alarm.poll.fd=sock;
@@ -754,7 +756,7 @@ int rhizome_server_http_response_header(rhizome_http_request *r, int result, con
 */
 int rhizome_server_http_send_bytes(rhizome_http_request *r)
 {
-  // keep writing until the write would block or we run out of data
+  // keep writing until we've written something or we run out of data
   while(r->request_type){
     
     /* Flush anything out of the buffer if present, before doing any further
@@ -782,10 +784,9 @@ int rhizome_server_http_send_bytes(rhizome_http_request *r)
 	  r->request_type&=~RHIZOME_HTTP_REQUEST_FROMBUFFER;
 	  r->buffer_offset=0; r->buffer_length=0;
 	}
-	
-	// go around the loop again to work out what we should do next
-	continue;
-	
+
+	// wait for another POLLOUT alarm
+	return 1;
       }
 
     switch(r->request_type&(~RHIZOME_HTTP_REQUEST_FROMBUFFER))

rhizome_store.c

@@ -17,6 +17,7 @@ int rhizome_exists(const char *fileHash){
 }
 
 int rhizome_open_write(struct rhizome_write *write, char *expectedFileHash, int64_t file_length, int priority){
+  write->blob_fd=-1;
   if (expectedFileHash){
     if (rhizome_exists(expectedFileHash))
       return 1;
@@ -65,7 +66,7 @@ int rhizome_open_write(struct rhizome_write *write, char *expectedFileHash, int6
       goto insert_row_fail;
     
     if (config.debug.externalblobs)
-      DEBUGF("Blob file created (fd=%d)", write->blob_fd);
+      DEBUGF("Writing to new blob file %s (fd=%d)", blob_path, write->blob_fd);
     
   }else{
     statement = sqlite_prepare(&retry,"INSERT OR REPLACE INTO FILEBLOBS(id,data) VALUES('%s',?)",write->id);
@@ -107,8 +108,11 @@ int rhizome_open_write(struct rhizome_write *write, char *expectedFileHash, int6
   }
   
   if (sqlite_exec_void_retry(&retry, "COMMIT;") == -1){
-    if (write->blob_fd>0){
+    if (write->blob_fd>=0){
+      if (config.debug.externalblobs)
+         DEBUGF("Cancel write to fd %d", write->blob_fd);
       close(write->blob_fd);
+      write->blob_fd=-1;
       unlink(blob_path);
     }
     return -1;
@@ -157,7 +161,8 @@ int rhizome_flush(struct rhizome_write *write_state){
       int r=write(write_state->blob_fd, write_state->buffer + ofs, write_state->data_size - ofs);
       if (r<0)
 	RETURN(WHY_perror("write"));
-      DEBUGF("Wrote %d bytes into external blob", r);
+      if (config.debug.externalblobs)
+        DEBUGF("Wrote %d bytes to fd %d", r, write_state->blob_fd);
       ofs+=r;
     }
   }else{
@@ -263,8 +268,11 @@ int rhizome_fail_write(struct rhizome_write *write){
     free(write->buffer);
   write->buffer=NULL;
   
-  if (write->blob_fd){
+  if (write->blob_fd>=0){
+    if (config.debug.externalblobs)
+      DEBUGF("Closing and removing fd %d", write->blob_fd);
     close(write->blob_fd);
+    write->blob_fd=-1;
     rhizome_store_delete(write->id);
   }
   
@@ -284,8 +292,12 @@ int rhizome_finish_write(struct rhizome_write *write){
     if (rhizome_flush(write))
       return -1;
   }
-  if (write->blob_fd)
+  if (write->blob_fd>=0){
+    if (config.debug.externalblobs)
+      DEBUGF("Closing fd %d", write->blob_fd);
     close(write->blob_fd);
+    write->blob_fd=-1;
+  }
   if (write->buffer)
     free(write->buffer);
   write->buffer=NULL;
@@ -482,6 +494,8 @@ int rhizome_open_read(struct rhizome_read *read, const char *fileid, int hash)
     }
     if ((read->length = lseek(read->blob_fd, 0, SEEK_END)) == -1)
       return WHYF_perror("lseek(%s,0,SEEK_END)", alloca_str_toprint(blob_path));
+    if (config.debug.externalblobs)
+      DEBUGF("Opened stored file %s as fd %d, len %llx",blob_path, read->blob_fd, read->length);
   }
   read->hash = hash;
   read->offset = 0;
@@ -497,12 +511,14 @@ int rhizome_read(struct rhizome_read *read_state, unsigned char *buffer, int buf
 {
   IN();
   int bytes_read = 0;
-  if (read_state->blob_fd != -1) {
+  if (read_state->blob_fd >= 0) {
     if (lseek(read_state->blob_fd, read_state->offset, SEEK_SET) == -1)
       RETURN(WHYF_perror("lseek(%d,%ld,SEEK_SET)", read_state->blob_fd, (long)read_state->offset));
     bytes_read = read(read_state->blob_fd, buffer, buffer_length);
     if (bytes_read == -1)
       RETURN(WHYF_perror("read(%d,%p,%ld)", read_state->blob_fd, buffer, (long)buffer_length));
+    if (config.debug.externalblobs)
+      DEBUGF("Read %d bytes from fd %d @%llx", bytes_read, read_state->blob_fd, read_state->offset);
   } else if (read_state->blob_rowid != -1) {
     sqlite_retry_state retry = SQLITE_RETRY_STATE_DEFAULT;
     do{
@@ -563,8 +579,11 @@ int rhizome_read(struct rhizome_read *read_state, unsigned char *buffer, int buf
 
 int rhizome_read_close(struct rhizome_read *read)
 {
-  if (read->blob_fd != -1)
+  if (read->blob_fd >=0){
+    if (config.debug.externalblobs)
+      DEBUGF("Closing store fd %d", read->blob_fd);
     close(read->blob_fd);
+  }
   read->blob_fd = -1;
   return 0;
 }

tests/rhizomeprotocol

@@ -247,7 +247,8 @@ setup_FileTransferBigMDPExtBlob() {
    foreach_instance +A +B \
       executeOk_servald config \
          set rhizome.http.enable 0 \
-         set rhizome.external_blobs 1
+         set rhizome.external_blobs 1 \
+	 set debug.externalblobs 1
    setup_bigfile_common
 }
 test_FileTransferBigMDPExtBlob() {
@@ -260,7 +261,8 @@ setup_FileTransferBigHTTPExtBlob() {
    foreach_instance +A +B \
       executeOk_servald config \
          set rhizome.mdp.enable 0 \
-         set rhizome.external_blobs 1
+         set rhizome.external_blobs 1 \
+	 set debug.externalblobs 1
    setup_bigfile_common
 }
 test_FileTransferBigHTTPExtBlob() {
@@ -321,7 +323,8 @@ setup_FileTransferMultiMDPExtBlob() {
    foreach_instance +A +B +C +D +E \
       executeOk_servald config \
          set rhizome.http.enable 0 \
-         set rhizome.external_blobs 1
+         set rhizome.external_blobs 1 \
+	 set debug.externalblobs 1
    setup_multitransfer_common
 }
 test_FileTransferMultiMDPExtBlob() {
@@ -334,7 +337,8 @@ setup_FileTransferMultiHTTPExtBlob() {
    foreach_instance +A +B +C +D +E \
       executeOk_servald config \
          set rhizome.mdp.enable 0 \
-         set rhizome.external_blobs 1
+         set rhizome.external_blobs 1 \
+	 set debug.externalblobs 1
    setup_multitransfer_common
 }
 test_FileTransferMultiHTTPExtBlob() {