Merge pull request #73 from Derecho-Project/use_zerocopy_delta_api

Adapt to the new IDeltaSupport API to save a copy during persistence

.github/workflows/build_use_zerocopy_delta_api.yml

@@ -0,0 +1,101 @@
+name: General Build Test
+
+on:
+  push:
+    branches: [ "use_zerocopy_delta_api" ]
+  pull_request:
+    branches: [ "use_zerocopy_delta_api" ]
+
+jobs:
+  build-check:
+    runs-on: ubuntu-latest
+    steps:
+    - run:  echo "The job is automatically triggered by a ${{github.event_name}} event."
+    - name: "Install APT packages"
+      run:  >
+            sudo apt update; 
+            sudo apt install libssl-dev librdmacm-dev libibverbs-dev libspdlog-dev -y;
+            sudo apt install libboost-all-dev ragel python3.10 python3-pip -y
+    - run:  g++ --version
+    - run:  cmake --version
+    - run:  lscpu
+    - run:  df -ha
+    - run:  cat /proc/meminfo
+    - name: Checkout Derecho code
+      uses: actions/checkout@v3
+      with:
+        repository: 'songweijia/derecho'
+        ref: 'master'
+        path: derecho
+    - run: echo "The Derecho HEAD repository has been clone to the runner."
+    - name: Checkout Cascade code
+      uses: actions/checkout@v3
+      with:
+        path: 'cascade'
+    - run:  echo "The ${{ github.repository }} repository has been cloned to the runner."
+    - name: Install libfabric
+      run:  ${{github.workspace}}/derecho/scripts/prerequisites/install-libfabric.sh ${{github.workspace}}/opt
+    - name: Install json library
+      run:  ${{github.workspace}}/derecho/scripts/prerequisites/install-json.sh ${{github.workspace}}/opt
+    - name: Install mutils
+      run:  ${{github.workspace}}/derecho/scripts/prerequisites/install-mutils.sh ${{github.workspace}}/opt
+    - name: Install mutils-containers
+      run:  ${{github.workspace}}/derecho/scripts/prerequisites/install-mutils-containers.sh ${{github.workspace}}/opt
+    - run:  mkdir ${{github.workspace}}/derecho/build-Release
+    - name: Build Derecho
+      run:  >
+            cd ${{github.workspace}}/derecho/build-Release;
+            export PREFIX=${{ github.workspace }}/opt;
+            export CMAKE_PREFIX_PATH=$PREFIX;
+            export C_INCLUDE_PATH=$PREFIX/include/;
+            export CPLUS_INCLUDE_PATH=$PREFIX/include/;
+            export LIBRARY_PATH=$PREFIX/lib/:$PREFIX/lib64/;
+            export LD_LIBRARY_PATH=$PREFIX/lib/:$PREFIX/lib64/;
+            cmake -DCMAKE_BUILD_TYPE=Release -DCMAKE_EXPORT_COMPILE_COMMANDS=ON -DCMAKE_INSTALL_PREFIX=${PREFIX} ..;
+            make -j2;
+            make install
+    - name: Install rpclib
+      run:  ${{github.workspace}}/cascade/scripts/prerequisites/install-rpclib.sh ${{github.workspace}}/opt
+    - name: Install hyperscan
+      run:  ${{github.workspace}}/cascade/scripts/prerequisites/install-hyperscan.sh ${{github.workspace}}/opt
+    - name: Install libwsong
+      run:  ${{github.workspace}}/cascade/scripts/prerequisites/install-libwsong.sh ${{github.workspace}}/opt
+    - name: Install python dependencies
+      run:  >
+            pip3 install -r ${{github.workspace}}/cascade/src/service/python/requirements.txt;
+            pip3 install -r ${{github.workspace}}/cascade/src/udl_zoo/python/requirements.txt
+    - run:  mkdir ${{github.workspace}}/cascade/build-Release
+    - name: Build Cascade
+      run:  >
+            cd ${{github.workspace}}/cascade/build-Release;
+            export PREFIX=${{github.workspace}}/opt;
+            export CMAKE_PREFIX_PATH=$PREFIX;
+            export C_INCLUDE_PATH=$PREFIX/include/;
+            export CPLUS_INCLUDE_PATH=$PREFIX/include/;
+            export LIBRARY_PATH=$PREFIX/lib/:$PREFIX/lib64/;
+            export LD_LIBRARY_PATH=$PREFIX/lib/:$PREFIX/lib64/;
+            export PATH=$PATH:$HOME/.local/bin/;
+            cmake -DCMAKE_BUILD_TYPE=Release -DCMAKE_EXPORT_COMPILE_COMMANDS=ON -DCMAKE_INSTALL_PREFIX=${PREFIX} -Dpybind11_DIR=`pybind11-config --cmakedir` ..;
+            make -j2;
+            make install;
+            cd ${{github.workspace}}/cascade/src/applications/standalone/console_printer_udl;
+            mkdir build;
+            cd build;
+            cmake ..;
+            make -j2;
+            cd ${{github.workspace}}/cascade/src/applications/standalone/dds;
+            mkdir build;
+            cd build;
+            cmake ..;
+            make -j2;
+            cd ${{github.workspace}}/cascade/src/applications/standalone/kvs_client;
+            mkdir build;
+            cd build;
+            cmake ..;
+            make -j2;
+            cd ${{github.workspace}}/cascade/src/applications/standalone/notification;
+            mkdir build;
+            cd build;
+            cmake ..;
+            make -j2
+    - run:  echo "The job's status is ${{job.status}}."

include/cascade/detail/delta_store_core.hpp

@@ -20,13 +20,6 @@ namespace cascade {
 template <typename KT, typename VT, KT* IK, VT* IV>
 class DeltaCascadeStoreCore : public mutils::ByteRepresentable,
                               public persistent::IDeltaSupport<DeltaCascadeStoreCore<KT, VT, IK, IV>> {
-    // TODO: use max payload size in subgroup configuration.
-#define DEFAULT_DELTA_BUFFER_CAPACITY (4096)
-    /**
-     * Initialize the delta data structure.
-     */
-    void initialize_delta();
-
 private:
 #if defined(__i386__) || defined(__x86_64__) || defined(_M_AMD64) || defined(_M_IX86)
     mutable std::atomic<persistent::version_t> lockless_v1;
@@ -36,39 +29,46 @@ class DeltaCascadeStoreCore : public mutils::ByteRepresentable,
 #endif
 
 public:
-    // delta
-    typedef struct {
-        size_t capacity;
-        size_t len;
-        uint8_t* buffer;
-        // methods
-        inline void set_data_len(const size_t& dlen);
-        inline uint8_t* data_ptr();
-        inline void calibrate(const size_t& dlen);
-        inline bool is_empty();
-        inline void clean();
-        inline void destroy();
-    } _Delta;
-    _Delta delta;
-
-    struct DeltaBytesFormat {
-        uint32_t op;
-        uint8_t first_data_byte;
+    /**
+     * @class DeltaType
+     * @brief the delta type that are stored in a delta.
+     * 1) The first sizeof(std::vector<KT>::size_type) bytes is the number of VT objects in the
+     * delta, followed by specified number of 
+     * 2) Serialized VT objects.
+     */
+    class DeltaType : public mutils::ByteRepresentable {
+    public:
+        /* The objects */
+        std::unordered_map<KT,VT> objects;
+        /** @fn DeltaType
+         *  @brief Constructor
+         */
+        DeltaType();
+        virtual std::size_t to_bytes(uint8_t*) const override;
+        virtual void post_object(const std::function<void(uint8_t const* const, std::size_t)>&) const override;
+        virtual std::size_t bytes_size() const override;
+        virtual void ensure_registered(mutils::DeserializationManager&);
+        static std::unique_ptr<DeltaType> from_bytes(mutils::DeserializationManager*,const uint8_t* const);
+        static mutils::context_ptr<DeltaType> from_bytes_noalloc(
+            mutils::DeserializationManager*,
+            const uint8_t* const);
+        static mutils::context_ptr<const DeltaType> from_bytes_noalloc_const(
+            mutils::DeserializationManager*,
+            const uint8_t* const);
     };
-
+    /** The delta is a list of keys for the objects that are changed by put or remove. */
+    std::vector<KT> delta;
+    /** The KV map */
     std::map<KT, VT> kv_map;
 
     //////////////////////////////////////////////////////////////////////////
-    // Delta is represented by an operation id and a list of
-    // argument. The operation id (OPID) is a 4 bytes integer.
-    // 1) put(const Object& object):
-    // [OPID:PUT]   [value]
-    // 2) remove(const KT& key)
-    // [OPID:REMOVE][key]
-    // 3) get(const KT& key)
-    // no need to prepare a delta
+    // Delta is represented by a list of objects for both put and remove
+    // operations, where the latter one is a list of objects with only key but
+    // is empty. Get operations will not create a delta. The first 4 bytes of
+    // the delta is the number of deltas.
     ///////////////////////////////////////////////////////////////////////////
-    virtual void finalizeCurrentDelta(const persistent::DeltaFinalizer& df) override;
+    virtual size_t currentDeltaSize() override;
+    virtual size_t currentDeltaToBytes(uint8_t * const buf, size_t buf_size) override;
     virtual void applyDelta(uint8_t const* const delta) override;
     static std::unique_ptr<DeltaCascadeStoreCore<KT, VT, IK, IV>> create(mutils::DeserializationManager* dm);
     /**