Ticket #1424: html5-zerocopy.patch

html5/js/Protocol.js

@@ -91 +91 @@
         this.cipher_out = null;
         this.mode = 'binary';   // Current WebSocket mode: 'binary', 'base64'
         this.rQ = [];                   // Receive queue
-        this.rQi = 0;                   // Receive queue index
-        this.rQmax = 10000;             // Max receive queue size before compacting
         this.sQ = [];                   // Send queue
         this.mQ = [];                   // Worker message queue
+        this.header = [];
 
         //Queue processing via intervals
         this.process_interval = 4;  //milliseconds
@@ -107 +106 @@
         var me = this;
         // init
         this.rQ          = [];
-        this.rQi                = 0;
         this.sQ          = [];
         this.websocket  = null;
         // connect the socket
@@ -124 +122 @@
         };
         this.websocket.onmessage = function (e) {
                 // push arraybuffer values onto the end
-                var u8 = new Uint8Array(e.data);
-                for (var i = 0; i < u8.length; i++) {
-                        me.rQ.push(u8[i]);
-                }
-                // wait for 8 bytes
-                //if (me.rQ.length >= 8) {
-                        //me._process();
-                //}
+                me.rQ.push(new Uint8Array(e.data));
         };
         this.start_processing();
 }
@@ -152 +143 @@
         var me = this;
         if(this.rQ_interval_id === null){
                 this.rQ_interval_id = setInterval(function(){
-                        if (me.rQ.length >= 8) {
+                        if (me.rQ.length > 0) {
                                 me.process_receive_queue();
                         }
                 }, this.process_interval);
@@ -185 +176 @@
         this.mQ_interval_id = null;
 }
 
+
 XpraProtocol.prototype.process_receive_queue = function() {
-        // peek at first 8 bytes of buffer
-        var buf = this._buffer_peek(8);
+        var i = 0, j = 0;
+        if (this.header.length<8 && this.rQ.length>0) {
+                //add from receive queue data to header until we get the 8 bytes we need:
+                while (this.header.length<8 && this.rQ.length>0) {
+                        var slice = this.rQ[0];
+                        var needed = 8-this.header.length;
+                        var n = Math.min(needed, slice.length);
+                        //console.log("header size", this.header.length, ", adding", n, "bytes from", slice.length);
+                        //copy at most n characters:
+                        for (i = 0; i < n; i++) {
+                                this.header.push(slice[i]);
+                        }
+                        if (slice.length>needed) {
+                                //replace the slice with what is left over:
+                                this.rQ[0] = slice.subarray(n);
+                        }
+                        else {
+                                //this slice has been fully consumed already:
+                                this.rQ.shift();
+                        }
+                }
 
-        if (buf[0] !== ord("P")) {
-                msg = "invalid packet header format: " + buf[0];
-                if (buf.length>1) {
-                        msg += ": ";
-                        for (c in buf) {
-                                msg += String.fromCharCode(c);
+                //verify the header format:
+                if (this.header[0] !== ord("P")) {
+                        msg = "invalid packet header format: " + this.header[0];
+                        if (this.header.length>1) {
+                                msg += ": ";
+                                for (c in this.header) {
+                                        msg += String.fromCharCode(c);
+                                }
                         }
+                        throw msg;
                 }
-                throw msg;
         }
 
-        var proto_flags = buf[1];
+        if (this.header.length<8) {
+                //we need more data to continue
+                return;
+        }
+
+        var proto_flags = this.header[1];
         var proto_crypto = proto_flags & 0x2;
-
         if (proto_flags!=0) {
                 // check for crypto protocol flag
                 if (!(proto_crypto)) {
@@ -210 +227 @@
                 }
         }
 
-        var level = buf[2];
-        var index = buf[3];
+        var level = this.header[2];
+        if (level & 0x20) {
+                throw "lzo compression is not supported";
+        }
+        var index = this.header[3];
+        if (index>=20) {
+                throw "invalid packet index: "+index;
+        }
         var packet_size = 0;
-        for (var i=0; i<4; i++) {
-                //debug("size header["+i+"]="+buf[4+i]);
+        for (i=0; i<4; i++) {
                 packet_size = packet_size*0x100;
-                packet_size += buf[4+i];
+                packet_size += this.header[4+i];
         }
+
         // work out padding if necessary
         var padding = 0
         if (proto_crypto) {
@@ -224 +247 @@
                 padding = (this.cipher_in_block_size - packet_size % this.cipher_in_block_size);
                 packet_size += padding;
         }
-        //debug("packet_size="+packet_size+", level="+level+", index="+index);
 
-        // wait for packet to be complete
-        // the header is still on the buffer so wait for packetsize+headersize bytes!
-        if (this.rQ.length < packet_size+8) {
-                // we already shifted the header off the buffer?
-                //debug("packet is not complete yet");
+        // verify that we have enough data for the full payload:
+        var rsize = 0;
+        for (i=0,j=this.rQ.length;i<j;++i) {
+                rsize += this.rQ[i].length;
+        }
+        if (rsize<packet_size) {
                 return;
         }
 
-        // packet is complete but header is still on buffer
-        this._buffer_shift(8);
-        //debug("got a full packet, shifting off "+packet_size);
-        var packet_data = this._buffer_shift(packet_size);
+        // done parsing the header, the next packet will need a new one:
+        this.header = []
 
+        var packet_data = null;
+        if (this.rQ[0].length==packet_size) {
+                //exact match: the payload is in a buffer already:
+                packet_data = this.rQ.shift();
+        }
+        else {
+                //aggregate all the buffers into "packet_data" until we get exactly "packet_size" bytes:
+                packet_data = new Uint8Array(packet_size);
+                rsize = 0;
+                while (rsize < packet_size) {
+                        var slice = this.rQ[0];
+                        var needed = packet_size - rsize;
+                        //console.log("slice:", slice.length, "bytes, needed", needed);
+                        if (slice.length>needed) {
+                                //add part of this slice:
+                                packet_data.set(slice.subarray(0, needed), rsize);
+                                rsize += needed;
+                                this.rQ[0] = slice.subarray(needed);
+                        }
+                        else {
+                                //add this slice in full:
+                                packet_data.set(slice, rsize);
+                                rsize += slice.length;
+                                this.rQ.shift();
+                        }
+                }
+        }
+
         // decrypt if needed
         if (proto_crypto) {
                 this.cipher_in.update(forge.util.createBuffer(uintToString(packet_data)));
                 var decrypted = this.cipher_in.output.getBytes();
                 packet_data = [];
-                for (var i=0; i<decrypted.length; i++)
+                for (i=0; i<decrypted.length; i++)
                         packet_data.push(decrypted[i].charCodeAt(0));
                 packet_data = packet_data.slice(0, -1 * padding);
         }
@@ -255 +304 @@
                         // lz4
                         // python-lz4 inserts the length of the uncompressed data as an int
                         // at the start of the stream
-                        var d = packet_data.splice(0, 4);
-                        // will always be little endian
+                        var d = packet_data.subarray(0, 4);
+                        // output buffer length is stored as little endian
                         var length = d[0] | (d[1] << 8) | (d[2] << 16) | (d[3] << 24);
                         // decode the LZ4 block
-                        var inflated = new Buffer(length);
-                        var uncompressedSize = LZ4.decodeBlock(packet_data, inflated);
-                        if(!proto_crypto)
-                                inflated = inflated.slice(0, uncompressedSize);
-                } else if (level & 0x20) {
-                        // lzo
+                        console.log("lz4 decompress packet size", packet_size, ", lz4 length=", length);
+                        var inflated = new Uint8Array(length);
+                        var uncompressedSize = LZ4.decodeBlock(packet_data, inflated, 4);
+                        // if lz4 errors out at the end of the buffer, ignore it:
+                        if (uncompressedSize<=0 && packet_size+uncompressedSize!=0) {
+                                console.error("failed to decompress lz4 data, error code:", uncompressedSize);
+                                return;
+                        }
                 } else {
                         // zlib
                         var inflated = new Zlib.Inflate(packet_data).decompress();
@@ -287 +338 @@
                         }
                         this.raw_packets = {}
                         // pass to our packet handler
-
                         if((packet[0] === 'draw') && (packet[6] !== 'scroll')){
                                 var img_data = packet[7];
                                 if (typeof img_data === 'string') {
                                         var uint = new Uint8Array(img_data.length);
-                                        for(var i=0,j=img_data.length;i<j;++i) {
+                                        for(i=0,j=img_data.length;i<j;++i) {
                                                 uint[i] = img_data.charCodeAt(i);
                                         }
                                         packet[7] = uint;
                                 }
-                                else {
-                                        packet[7] = new Uint8Array(packet[7]);
-                                }
                         }
-                        if(this.is_worker){
+                        if (this.is_worker){
                                 this.mQ[this.mQ.length] = packet;
                         } else {
                                 this.packet_handler(packet, this.packet_ctx);
@@ -314 +361 @@
         }
 
         // see if buffer still has unread packets
-        if (this.rQ.length >= 8) {
+        if (this.rQ.length > 0) {
                 this.process_receive_queue();
         }
 }
@@ -405 +452 @@
         this.cipher_out.start({iv: caps['cipher.iv']});
 }
 
-XpraProtocol.prototype._buffer_peek = function(bytes) {
-        return this.rQ.slice(0, 0+bytes);
-}
 
-XpraProtocol.prototype._buffer_shift = function(bytes) {
-        return this.rQ.splice(0, 0+bytes);;
-}
-
-
 /*
 If we are in a web worker, set up an instance of the protocol
 */