Ticket #348: compressed_cc.diff

xapian-core/backends/brass/brass_databasereplicator.cc

@@ -20 +20 @@
  * USA
  */
 
+#include <iostream>
 #include <config.h>
 
 #include "brass_databasereplicator.h"
@@ -34 +35 @@
 #include "debuglog.h"
 #include "fd.h"
 #include "filetests.h"
+#include "internaltypes.h"
 #include "io_utils.h"
 #include "pack.h"
 #include "net/remoteconnection.h"
@@ -42 +44 @@
 #include "str.h"
 #include "stringutils.h"
 
+
 #ifdef __WIN32__
 # include "msvc_posix_wrapper.h"
 #endif
@@ -52 +55 @@
 using namespace Xapian;
 
 BrassDatabaseReplicator::BrassDatabaseReplicator(const string & db_dir_)
-        : db_dir(db_dir_)
+    : db_dir(db_dir_),
+      comp_stream(Z_DEFAULT_STRATEGY)
 {
 }
 
@@ -191 +195 @@
     }
     {
         FD closer(fd);
-
+        Bytef out[8192];
         while (true) {
             conn.get_message_chunk(buf, REASONABLE_CHANGESET_SIZE, end_time);
             ptr = buf.data();
             end = ptr + buf.size();
-
             uint4 block_number;
             if (!unpack_uint(&ptr, end, &block_number))
                 throw NetworkError("Invalid block number in changeset");
@@ -205 +208 @@
                 break;
             --block_number;
 
-            conn.get_message_chunk(buf, changeset_blocksize, end_time);
+            conn.get_message_chunk(buf, REASONABLE_CHANGESET_SIZE, end_time);
+            ptr = buf.data();
+            end = ptr + buf.size();
+            unsigned int compressed_block_size;
+            if(!unpack_uint(&ptr, end, &compressed_block_size))
+                throw NetworkError("Invalid v2 cphangeset");
+            buf.erase(0, ptr - buf.data());
+
+            if (compressed_block_size > 0) {
+                string ubuf;
+                ubuf.reserve(changeset_blocksize);
+                conn.get_message_chunk(buf, changeset_blocksize, end_time);
+                comp_stream.lazy_alloc_inflate_zstream();
+                comp_stream.inflate_zstream->next_in = (Bytef*)const_cast<char *>(buf.data());
+                comp_stream.inflate_zstream->avail_in = (uInt)(buf.size());
+                comp_stream.inflate_zstream->next_out = out;
+                comp_stream.inflate_zstream->avail_out = (uInt)sizeof(out);
+                int err = inflate(comp_stream.inflate_zstream, Z_FINISH);
+                if (err == Z_STREAM_END) {
+                    std::cout << comp_stream.inflate_zstream->next_out - out << std::endl;
+                    std::cout << out << std::endl;
+                    ubuf.append(reinterpret_cast<const char *>(out),
+                                comp_stream.inflate_zstream->next_out - out);
+                    swap(buf, ubuf);
+                }
+            } else {
+                conn.get_message_chunk(buf, changeset_blocksize, end_time);
+            }
             if (buf.size() < changeset_blocksize)
                 throw NetworkError("Incomplete block in changeset");
 
@@ -255 +285 @@
     buf.erase(0, 12);
     const char *ptr = buf.data();
     const char *end = ptr + buf.size();
-
     unsigned int changes_version;
     if (!unpack_uint(&ptr, end, &changes_version))
         throw NetworkError("Couldn't read a valid version number from changeset");

xapian-core/backends/brass/brass_databasereplicator.h

@@ -24 +24 @@
 #define XAPIAN_INCLUDED_BRASS_DATABASEREPLICATOR_H
 
 #include "backends/databasereplicator.h"
+#include "zlib_utils.h"
 
 class BrassDatabaseReplicator : public Xapian::DatabaseReplicator {
     private:
@@ -31 +32 @@
          */
         std::string db_dir;
 
+        CompressionStream comp_stream;
+
         /** Process a chunk which holds a base block.
          */
         void process_changeset_chunk_base(const std::string & tablename,

xapian-core/backends/brass/brass_replicate_internal.h

@@ -28 +28 @@
 
 // The current version of changeset files.
 // 1  - initial implementation
-#define CHANGES_VERSION 1u
+// 2  - compressed changesets
+#define CHANGES_VERSION 2u
 
 // Must be big enough to ensure that the start of the changeset (up to the new
 // revision number) will fit in this much space.

xapian-core/backends/brass/brass_table.cc

@@ -1818 +1818 @@
     pack_string(buf, tablename);
     pack_uint(buf, block_size);
 
-    if (compressed) { 
-        comp_stream.lazy_alloc_deflate_zstream();
-
-        comp_stream.deflate_zstream->next_in = (Bytef *)const_cast<char *>(buf.data());
-        comp_stream.deflate_zstream->avail_in = (uInt)buf.size();
-        int err = deflate(comp_stream.deflate_zstream, Z_FINISH);
-        if (err == Z_STREAM_END) {
-            io_write(changes_fd, reinterpret_cast<const char *>(buf.data()),
-                     comp_stream.deflate_zstream->total_out);
-        } else {
-            // The deflate failed, try to write data uncompressed
-            io_write(changes_fd, buf.data(), buf.size());
-        }
-    }
-    else {
-        io_write(changes_fd, buf.data(), buf.size());
-    }
+    // Write the table name and block size to the file
+    io_write(changes_fd, buf.data(), buf.size());
         
     // Compare the old and new bitmaps to find blocks which have changed, and
     // write them to the file descriptor.
@@ -1845 +1830 @@
         while (base.find_changed_block(&n)) {
             buf.resize(0);
             pack_uint(buf, n + 1);
-            if (compressed) { 
-                comp_stream.lazy_alloc_deflate_zstream();
-
-                comp_stream.deflate_zstream->next_in = (Bytef *)const_cast<char *>(buf.data());
-                comp_stream.deflate_zstream->avail_in = (uInt)buf.size();
-                int err = deflate(comp_stream.deflate_zstream, Z_FINISH);
-                if (err == Z_STREAM_END) {
-                    io_write(changes_fd, reinterpret_cast<const char *>(buf.data()),
-                             comp_stream.deflate_zstream->total_out);
-                } else {
-                    // The deflate failed, try to write data uncompressed
-                    io_write(changes_fd, buf.data(), buf.size());
-                }
-            }
-            else {
-                io_write(changes_fd, buf.data(), buf.size());
-            }
+            // Write the block number to the file
+            io_write(changes_fd, buf.data(), buf.size());
 
             // Read block n.
             read_block(n, p);
@@ -1869 +1839 @@
             // Write block n to the file.
             if (compressed) { 
                 comp_stream.lazy_alloc_deflate_zstream();
-
-                comp_stream.deflate_zstream->next_in = (Bytef *)(p);
-                comp_stream.deflate_zstream->avail_in = (uInt)block_size;
-                int err = deflate(comp_stream.deflate_zstream, Z_FINISH);
-                if (err == Z_STREAM_END) {
-                    io_write(changes_fd, reinterpret_cast<const char *>(p),
+                comp_stream.compress(p, block_size);
+                if (comp_stream.zerr == Z_STREAM_END) {
+                    buf.resize(0);
+                    pack_uint(buf, comp_stream.deflate_zstream->total_out);
+                    io_write(changes_fd, buf.data(), buf.size());
+                    io_write(changes_fd, reinterpret_cast<const char *>(comp_stream.out),
                              comp_stream.deflate_zstream->total_out);
                 } else {
                     // The deflate failed, try to write data uncompressed
+                    buf.resize(0);
+                    pack_uint(buf, 0u);
+                    io_write(changes_fd, buf.data(), buf.size());
                     io_write(changes_fd, reinterpret_cast<const char *>(p), block_size);
                 }
             }
             else {
+                buf.resize(0);
+                pack_uint(buf, 0u);
+                io_write(changes_fd, buf.data(), buf.size());
                 io_write(changes_fd, reinterpret_cast<const char *>(p), block_size);
             }
             ++n;
@@ -1894 +1870 @@
     }
     buf.resize(0);
     pack_uint(buf, 0u);
-
-    if (compressed) { 
-        comp_stream.lazy_alloc_deflate_zstream();
-
-        comp_stream.deflate_zstream->next_in = (Bytef *)const_cast<char *>(buf.data());
-        comp_stream.deflate_zstream->avail_in = (uInt)buf.size();
-        int err = deflate(comp_stream.deflate_zstream, Z_FINISH);
-        if (err == Z_STREAM_END) {
-            io_write(changes_fd, reinterpret_cast<const char *>(buf.data()),
-                     comp_stream.deflate_zstream->total_out);
-        } else {
-            // The deflate failed, try to write data uncompressed
-            io_write(changes_fd, buf.data(), buf.size());
-        }
-    }
-    else {
-        io_write(changes_fd, buf.data(), buf.size());
-    }
+    // Write 0 for end of blocks
+    io_write(changes_fd, buf.data(), buf.size());
 }
 
 void

xapian-core/backends/brass/brass_table.h

@@ -41 +41 @@
 #include <algorithm>
 #include <string>
 
-#include <zlib.h>
-
 #define DONT_COMPRESS -1
 
 /** Even for items of at maximum size, it must be possible to get this number of

xapian-core/common/Makefile.mk

@@ -38 +38 @@
         common/stringutils.h\
         common/submatch.h\
         common/unaligned.h\
-        common/utils.h\
-        common/valuelist.h\
-        common/valuestats.h\
-        common/vectortermlist.h\
-        common/weightinternal.h\
         common/zlib_utils.h
         common/unaligned.h
 
@@ -67 +62 @@
         common/serialise-double.cc\
         common/socket_utils.cc\
         common/str.cc\
-        common/stringutils.cc
+        common/stringutils.cc\
         common/zlib_utils.cc
 
 

xapian-core/common/zlib_utils.cc

@@ -6 +6 @@
 
 CompressionStream::CompressionStream(int compress_strategy_)
     : compress_strategy(compress_strategy_),
+      zerr(0),
+      out_len(0),
+      out(NULL),
       deflate_zstream(NULL),
       inflate_zstream(NULL)
 {
@@ -26 +29 @@
         (void) inflateEnd(inflate_zstream);
         delete inflate_zstream;
     }
+    
+    if (out) {
+        delete [] out;
+    }
+}
+
+
+void
+CompressionStream::compress(string & buf) {
+    out_len = buf.size() - 1;
+    out = new unsigned char[out_len];
+    deflate_zstream->avail_in = (uInt)buf.size();
+    deflate_zstream->next_in = (Bytef *)const_cast<char *>(buf.data());
+    deflate_zstream->next_out = out;
+    deflate_zstream->avail_out = (uInt)out_len;
+    zerr = deflate(deflate_zstream, Z_FINISH);
 }
 
+
+void
+CompressionStream::compress(byte * buf, int size) {
+    out_len = size - 1;
+    out = new unsigned char[out_len];
+    deflate_zstream->avail_in = (uInt)size;
+    deflate_zstream->next_in = (Bytef *)(buf);
+    deflate_zstream->next_out = out;
+    deflate_zstream->avail_out = (uInt)out_len;
+    zerr = deflate(deflate_zstream, Z_FINISH);
+}
+
+// void
+// CompressionStream::decompress(string & buf) {
+//     inflate_zstream->next_in = (Bytef*)const_cast<char *>(tag->data());
+//     inflate_zstream->avail_in = (uInt)tag->size();
+//     int err = Z_OK;
+//     while (err != Z_STREAM_END) {
+//      comp_stream.inflate_zstream->next_out = buf;
+//      comp_stream.inflate_zstream->avail_out = (uInt)sizeof(buf);
+//      err = inflate(comp_stream.inflate_zstream, Z_SYNC_FLUSH);
+//      if (err == Z_BUF_ERROR && comp_stream.inflate_zstream->avail_in == 0) {
+//          LOGLINE(DB, "Z_BUF_ERROR - faking checksum of " << comp_stream.inflate_zstream->adler);
+//          Bytef header2[4];
+//          setint4(header2, 0, comp_stream.inflate_zstream->adler);
+//          comp_stream.inflate_zstream->next_in = header2;
+//          comp_stream.inflate_zstream->avail_in = 4;
+//          err = inflate(comp_stream.inflate_zstream, Z_SYNC_FLUSH);
+//          if (err == Z_STREAM_END) break;
+//      }
+
+//      if (err != Z_OK && err != Z_STREAM_END) {
+//          if (err == Z_MEM_ERROR) throw std::bad_alloc();
+//          string msg = "inflate failed";
+//          if (comp_stream.inflate_zstream->msg) {
+//              msg += " (";
+//              msg += comp_stream.inflate_zstream->msg;
+//              msg += ')';
+//          }
+//          throw Xapian::DatabaseError(msg);
+//      }
+
+//      utag.append(reinterpret_cast<const char *>(buf),
+//                  comp_stream.inflate_zstream->next_out - buf);
+//     }
+//     if (utag.size() != comp_stream.inflate_zstream->total_out) {
+//      string msg = "compressed tag didn't expand to the expected size: ";
+//      msg += str(utag.size());
+//      msg += " != ";
+//      // OpenBSD's zlib.h uses off_t instead of uLong for total_out.
+//      msg += str((size_t)comp_stream.inflate_zstream->total_out);
+//      throw Xapian::DatabaseCorruptError(msg);
+//     }
+// }
+
+
 void
 CompressionStream::lazy_alloc_deflate_zstream() const {
     if (usual(deflate_zstream)) {
@@ -102 +177 @@
         throw Xapian::DatabaseError(msg);
     }
 }
-
-// void
-// BrassTable::lazy_alloc_deflate_zstream() const {
-//     if (usual(deflate_zstream)) {
-//      if (usual(deflateReset(deflate_zstream) == Z_OK)) return;
-//      // Try to recover by deleting the stream and starting from scratch.
-//      delete deflate_zstream;
-//     }
-
-//     deflate_zstream = new z_stream;
-
-//     deflate_zstream->zalloc = reinterpret_cast<alloc_func>(0);
-//     deflate_zstream->zfree = reinterpret_cast<free_func>(0);
-//     deflate_zstream->opaque = (voidpf)0;
-
-//     // -15 means raw deflate with 32K LZ77 window (largest)
-//     // memLevel 9 is the highest (8 is default)
-//     int err;
-//     err = deflateInit2(deflate_zstream, Z_DEFAULT_COMPRESSION, Z_DEFLATED,
-//                     -15, 9, compress_strategy);
-//     if (rare(err != Z_OK)) {
-//      if (err == Z_MEM_ERROR) {
-//          delete deflate_zstream;
-//          deflate_zstream = 0;
-//          throw std::bad_alloc();
-//      }
-//      string msg = "deflateInit2 failed (";
-//      if (deflate_zstream->msg) {
-//          msg += deflate_zstream->msg;
-//      } else {
-//          msg += str(err);
-//      }
-//      msg += ')';
-//      delete deflate_zstream;
-//      deflate_zstream = 0;
-//      throw Xapian::DatabaseError(msg);
-//     }
-// }
-
-// void
-// BrassTable::lazy_alloc_inflate_zstream() const {
-//     if (usual(inflate_zstream)) {
-//      if (usual(inflateReset(inflate_zstream) == Z_OK)) return;
-//      // Try to recover by deleting the stream and starting from scratch.
-//      delete inflate_zstream;
-//     }
-
-//     inflate_zstream = new z_stream;
-
-//     inflate_zstream->zalloc = reinterpret_cast<alloc_func>(0);
-//     inflate_zstream->zfree = reinterpret_cast<free_func>(0);
-
-//     inflate_zstream->next_in = Z_NULL;
-//     inflate_zstream->avail_in = 0;
-
-//     int err = inflateInit2(inflate_zstream, -15);
-//     if (rare(err != Z_OK)) {
-//      if (err == Z_MEM_ERROR) {
-//          delete inflate_zstream;
-//          inflate_zstream = 0;
-//          throw std::bad_alloc();
-//      }
-//      string msg = "inflateInit2 failed (";
-//      if (inflate_zstream->msg) {
-//          msg += inflate_zstream->msg;
-//      } else {
-//          msg += str(err);
-//      }
-//      msg += ')';
-//      delete inflate_zstream;
-//      inflate_zstream = 0;
-//      throw Xapian::DatabaseError(msg);
-//     }
-// }

xapian-core/common/zlib_utils.h

@@ +1 @@
+#ifndef XAPIAN_INCLUDED_ZLIB_UTILS_H
+#define XAPIAN_INCLUDED_ZLIB_UTILS_H
 
-#include <zlib.h>
+#include "debuglog.h"
+
+#include "internaltypes.h"
 
 #include "xapian/error.h"
 
-#include "debuglog.h"
-#include "utils.h"
+#include <zlib.h>
 
 using namespace std;
 
@@ -19 +22 @@
 
     int compress_strategy;
 
+    int zerr;
+
+    unsigned long out_len;
+    
+    unsigned char * out;
+
     /// Zlib state object for deflating
     mutable z_stream *deflate_zstream;
 
@@ -30 +39 @@
 
     /// Allocate the zstream for inflating, if not already allocated.
     void lazy_alloc_inflate_zstream() const;
+
+    void compress(string &);
+    void compress(byte *, int);
 };
+
+#endif // XAPIAN_INCLUDED_ZLIB_UTILS_H

xapian-core/tests/api_query.cc

@@ -66 +66 @@
     Xapian::Query q;
     q = Xapian::Query("foo") & Xapian::Query("bar");
     TEST_STRINGS_EQUAL(q.get_description(), "Query((foo AND bar))");
-    q = Xapian::Query("foo") &~ Xapian::Query("bar");
+    q = Xapian::Query("foo") & (~Xapian::Query("bar"));
     TEST_STRINGS_EQUAL(q.get_description(), "Query((foo AND_NOT bar))");
     q = ~Xapian::Query("bar");
     TEST_STRINGS_EQUAL(q.get_description(), "Query((<alldocuments> AND_NOT bar))");