Ticket #185: Upload_with_status_queue.patch

source/ch/ethz/ssh2/sftp/SFTPv3Client.java

@@ -1092 +1092 @@
      * A read  is divided into multiple requests sent sequentially before
      * reading any status from the server
      */
-    private static class OutstandingRequest {
+    private static class OutstandingReadRequest {
         int req_id;
         /**
          * Read offset to request on server starting at the file offset for the first request.
@@ -1135 +1135 @@
     /**
      * @param parallelism
      */
-    public void setDownloadRequestParallelism(int parallelism) {
+    public void setRequestParallelism(int parallelism) {
         this.parallelism = Math.min(parallelism, DEFAULT_MAX_PARALLELISM);
         log.log("setDownloadRequestParallelism:" + this.parallelism);
     }
@@ -1143 +1143 @@
     /**
      * Mapping request ID to request.
      */
-    Map<Integer, OutstandingRequest> pendingQueue
-            = new HashMap<Integer, OutstandingRequest>();
+    Map<Integer, OutstandingReadRequest> pendingReadQueue
+            = new HashMap<Integer, OutstandingReadRequest>();
 
     /**
      * Read bytes from a file in a parallel fashion. As many bytes as you want will be read.
@@ -1177 +1177 @@
         int clientOffset = dstoff;
 
         long serverOffset = fileOffset;
-        for(OutstandingRequest r : pendingQueue.values()) {
+        for(OutstandingReadRequest r : pendingReadQueue.values()) {
             // Server offset should take pending requests into account.
             serverOffset += r.len;
         }
 
         while(true) {
-            /* If there was an error and no outstanding request - stop */
-            if((pendingQueue.size() == 0) && errorOccured) {
+            // Stop if there was an error and no outstanding request
+            if((pendingReadQueue.size() == 0) && errorOccured) {
                 break;
             }
 
-            /* Send as many requests as we are allowed to */
-            while(pendingQueue.size() < parallelism) {
+            // Send as many requests as we are allowed to
+            while(pendingReadQueue.size() < parallelism) {
                 if(errorOccured) {
                     break;
                 }
-                /* Send the next request */
-                OutstandingRequest req = new OutstandingRequest();
+                // Send the next read request
+                OutstandingReadRequest req = new OutstandingReadRequest();
                 req.req_id = generateNextRequestID();
                 req.serverOffset = serverOffset;
                 req.len = (remaining > len) ? len : remaining;
@@ -1207 +1207 @@
 
                 sendReadRequest(req.req_id, handle, req.serverOffset, req.len);
 
-                pendingQueue.put(req.req_id, req);
+                pendingReadQueue.put(req.req_id, req);
             }
-
-            /* Are we done? */
-            if(pendingQueue.size() == 0) {
+            if(pendingReadQueue.size() == 0) {
                 break;
             }
 
-            /* No, receive a single answer */
+            // Receive a single answer
             byte[] resp = receiveMessage(34000);
-
             TypesReader tr = new TypesReader(resp);
-
             int type = tr.readByte();
-            int rep_id = tr.readUINT32();
-
-            /* Search the pending queue */
-            OutstandingRequest req = pendingQueue.get(rep_id);
-
-            /* Should shutdown here, no point in going on */
+            // Search the pending queue
+            OutstandingReadRequest req = pendingReadQueue.remove(tr.readUINT32());
             if(null == req) {
                 throw new IOException("The server sent an invalid id field.");
             }
-
-            pendingQueue.remove(rep_id);
-
-            /* Evaluate the answer */
+            // Evaluate the answer
             if(type == Packet.SSH_FXP_STATUS) {
                 /* In any case, stop sending more packets */
 
@@ -1246 +1235 @@
                 }
                 // Flag to read all pending requests but don't send any more.
                 errorOccured = true;
-                if(pendingQueue.isEmpty()) {
+                if(pendingReadQueue.isEmpty()) {
                     if(ErrorCodes.SSH_FX_EOF == code) {
                         return -1;
                     }
@@ -1254 +1243 @@
                 }
             }
             else if(type == Packet.SSH_FXP_DATA) {
-                /* OK, collect data */
+                // OK, collect data
                 int readLen = tr.readUINT32();
 
                 if((readLen < 0) || (readLen > req.len)) {
@@ -1278 +1267 @@
                     log.log("Requesting again: " + req.serverOffset + "/" + req.len);
                     sendReadRequest(req.req_id, handle, req.serverOffset, req.len);
 
-                    pendingQueue.put(req.req_id, req);
+                    pendingReadQueue.put(req.req_id, req);
                 }
                 return readLen;
             }
@@ -1291 +1280 @@
     }
 
     /**
+     * A read  is divided into multiple requests sent sequentially before
+     * reading any status from the server
+     */
+    private static class OutstandingStatusRequest {
+        int req_id;
+    }
+
+    /**
+     * Mapping request ID to request.
+     */
+    Map<Integer, OutstandingStatusRequest> pendingStatusQueue
+            = new HashMap<Integer, OutstandingStatusRequest>();
+
+    /**
      * Write bytes to a file. If <code>len</code> &gt; 32768, then the write operation will
      * be split into multiple writes.
      *
@@ -1304 +1307 @@
     public void upload(SFTPv3FileHandle handle, long fileOffset, byte[] src, int srcoff, int len) throws IOException {
         checkHandleValidAndOpen(handle);
 
-        while(len > 0) {
-            int writeRequestLen = len;
+        // Send the next write request
+        OutstandingStatusRequest req = new OutstandingStatusRequest();
+        req.req_id = generateNextRequestID();
 
-            if(writeRequestLen > 32768) {
-                writeRequestLen = 32768;
-            }
-
-            int req_id = generateNextRequestID();
-
-            TypesWriter tw = new TypesWriter();
-            tw.writeString(handle.fileHandle, 0, handle.fileHandle.length);
-            tw.writeUINT64(fileOffset);
-            tw.writeString(src, srcoff, writeRequestLen);
+        TypesWriter tw = new TypesWriter();
+        tw.writeString(handle.fileHandle, 0, handle.fileHandle.length);
+        tw.writeUINT64(fileOffset);
+        tw.writeString(src, srcoff, len);
 
-            log.log("Sending SSH_FXP_WRITE...");
-            sendMessage(Packet.SSH_FXP_WRITE, req_id, tw.getBytes());
+        log.log("Sending SSH_FXP_WRITE...");
+        sendMessage(Packet.SSH_FXP_WRITE, req.req_id, tw.getBytes());
 
-            fileOffset += writeRequestLen;
+        pendingStatusQueue.put(req.req_id, req);
 
-            srcoff += writeRequestLen;
-            len -= writeRequestLen;
+        // Only read next status if parallelism reached
+        while(pendingStatusQueue.size() >= parallelism) {
+            this.readStatus();
+        }
+    }
 
-            byte[] resp = receiveMessage(34000);
+    private void readStatus() throws IOException {
+        byte[] resp = receiveMessage(34000);
 
-            TypesReader tr = new TypesReader(resp);
+        TypesReader tr = new TypesReader(resp);
+        int type = tr.readByte();
 
-            int t = tr.readByte();
-
-            int rep_id = tr.readUINT32();
-            if(rep_id != req_id) {
-                throw new IOException("The server sent an invalid id field.");
-            }
+        // Search the pending queue
+        OutstandingStatusRequest status = pendingStatusQueue.remove(tr.readUINT32());
+        if(null == status) {
+            throw new IOException("The server sent an invalid id field.");
+        }
 
-            if(t != Packet.SSH_FXP_STATUS) {
-                throw new IOException("The SFTP server sent an unexpected packet type (" + t + ")");
+        // Evaluate the answer
+        if(type == Packet.SSH_FXP_STATUS) {
+            // In any case, stop sending more packets
+            int code = tr.readUINT32();
+            if(log.isEnabled()) {
+                String[] desc = ErrorCodes.getDescription(code);
+                log.log("Got SSH_FXP_STATUS (" + status.req_id + ") (" + ((desc != null) ? desc[0] : "UNKNOWN") + ")");
             }
-
-            int errorCode = tr.readUINT32();
-
-            if(errorCode == ErrorCodes.SSH_FX_OK) {
-                continue;
+            if(code == ErrorCodes.SSH_FX_OK) {
+                return;
             }
-
-            String errorMessage = tr.readString();
-
-            throw new SFTPException(errorMessage, errorCode);
+            String msg = tr.readString();
+            throw new SFTPException(msg, code);
         }
+        throw new IOException("The SFTP server sent an unexpected packet type (" + type + ")");
     }
 
     /**
@@ -1360 +1363 @@
      * @throws IOException
      */
     public void closeFile(SFTPv3FileHandle handle) throws IOException {
-        if(handle == null) {
-            throw new IllegalArgumentException("the handle argument may not be null");
-        }
         try {
+            while(!pendingStatusQueue.isEmpty()) {
+                this.readStatus();
+            }
             if(!handle.isClosed) {
                 closeHandle(handle.fileHandle);
             }