[WIP] ARROW-4418 [Plasma] replace event loop with boost::asio for plasma store

LICENSE.txt

@@ -221,37 +221,6 @@ limitations under the License.
 
 --------------------------------------------------------------------------------
 
-src/plasma/thirdparty/ae: Modified / 3-Clause BSD
-
-Copyright (c) 2006-2010, Salvatore Sanfilippo <antirez at gmail dot com>
-All rights reserved.
-
-Redistribution and use in source and binary forms, with or without
-modification, are permitted provided that the following conditions are met:
-
- * Redistributions of source code must retain the above copyright notice,
-   this list of conditions and the following disclaimer.
- * Redistributions in binary form must reproduce the above copyright
-   notice, this list of conditions and the following disclaimer in the
-   documentation and/or other materials provided with the distribution.
- * Neither the name of Redis nor the names of its contributors may be used
-   to endorse or promote products derived from this software without
-   specific prior written permission.
-
-THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
-AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
-IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
-ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
-LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
-CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
-SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
-INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
-CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
-ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
-POSSIBILITY OF SUCH DAMAGE.
-
---------------------------------------------------------------------------------
-
 src/plasma/thirdparty/dlmalloc.c: CC0
 
 This is a version (aka dlmalloc) of malloc/free/realloc written by

cpp/apidoc/tutorials/plasma.md

@@ -384,16 +384,14 @@ sealed in the object store. This may especially be handy when your
 program is collaborating with other Plasma clients, and needs to know
 when they make objects available.
 
-First, you can subscribe your current Plasma client to such notifications
-by getting a file descriptor:
+First, you can subscribe your current Plasma client to such notifications:
 
 ```cpp
 // Start receiving notifications into file_descriptor.
-int fd;
-ARROW_CHECK_OK(client.Subscribe(&fd));
+ARROW_CHECK_OK(client.Subscribe());
 ```
 
-Once you have the file descriptor, you can have your current Plasma client
+Once you have subscribed, you can have your current Plasma client
 wait to receive the next object notification. Object notifications
 include information such as Object ID, data size, and metadata size of
 the next newly available object:
@@ -404,7 +402,7 @@ the next newly available object:
 ObjectID object_id;
 int64_t data_size;
 int64_t metadata_size;
-ARROW_CHECK_OK(client.GetNotification(fd, &object_id, &data_size, &metadata_size));
+ARROW_CHECK_OK(client.GetNotification(&object_id, &data_size, &metadata_size));
 
 // Get the newly available object.
 ObjectBuffer object_buffer;
@@ -423,14 +421,13 @@ int main(int argc, char** argv) {
   PlasmaClient client;
   ARROW_CHECK_OK(client.Connect("/tmp/plasma"));
 
-  int fd;
-  ARROW_CHECK_OK(client.Subscribe(&fd));
+  ARROW_CHECK_OK(client.Subscribe());
 
   ObjectID object_id;
   int64_t data_size;
   int64_t metadata_size;
   while (true) {
-    ARROW_CHECK_OK(client.GetNotification(fd, &object_id, &data_size, &metadata_size));
+    ARROW_CHECK_OK(client.GetNotification(&object_id, &data_size, &metadata_size));
 
     std::cout << "Received object notification for object_id = "
               << object_id.hex() << ", with data_size = " << data_size

cpp/src/plasma/CMakeLists.txt

@@ -22,6 +22,7 @@ add_custom_target(plasma-tests)
 add_dependencies(plasma-all plasma plasma-tests plasma-benchmarks)
 
 # For the moment, Plasma is versioned like Arrow
+project(plasma VERSION "${ARROW_BASE_VERSION}")
 set(PLASMA_VERSION "${ARROW_VERSION}")
 
 find_package(Threads)
@@ -32,25 +33,61 @@ set(PLASMA_FULL_SO_VERSION "${ARROW_FULL_SO_VERSION}")
 
 set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wno-conversion")
 
+# Compile flatbuffers
+
+set(PLASMA_FBS_SRC "${CMAKE_CURRENT_LIST_DIR}/format/plasma.fbs"
+                   "${CMAKE_CURRENT_LIST_DIR}/format/common.fbs")
+set(OUTPUT_DIR ${ARROW_BINARY_DIR}/src/plasma)
+
+set(PLASMA_FBS_OUTPUT_FILES "${OUTPUT_DIR}/common_generated.h"
+                            "${OUTPUT_DIR}/plasma_generated.h")
+
+add_custom_target(gen_plasma_fbs DEPENDS ${PLASMA_FBS_OUTPUT_FILES})
+
+add_dependencies(gen_plasma_fbs flatbuffers::flatc)
+
+add_custom_command(
+  OUTPUT ${PLASMA_FBS_OUTPUT_FILES}
+         # The --gen-object-api flag generates a C++ class MessageT for each
+         # flatbuffers message Message, which can be used to store deserialized
+         # messages in data structures. This is currently used for ObjectInfo for
+         # example.
+  COMMAND flatbuffers::flatc
+          -c
+          -o
+          ${OUTPUT_DIR}
+          ${PLASMA_FBS_SRC}
+          --gen-object-api
+          --scoped-enums
+  DEPENDS ${PLASMA_FBS_SRC}
+  COMMENT "Running flatc compiler on ${PLASMA_FBS_SRC}"
+  VERBATIM)
+
 set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fPIC")
 
+# Set compiling options for asio headers.
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-conversion -Wno-documentation")
+
+set(PLASMA_IO_SRCS
+    fling.cc
+    io/basic_connection.cc
+    io/connection.cc
+    protocol.cc)
+
 set(PLASMA_SRCS
     client.cc
     common.cc
-    fling.cc
-    io.cc
     malloc.cc
     plasma.cc
-    protocol.cc)
+    protocol.cc
+    ${PLASMA_IO_SRCS})
 
 set(PLASMA_STORE_SRCS
     dlmalloc.cc
-    events.cc
     eviction_policy.cc
-    quota_aware_policy.cc
     plasma_allocator.cc
-    store.cc
-    thirdparty/ae/ae.c)
+    ${PLASMA_IO_SRCS}
+    store.cc)
 
 set(PLASMA_LINK_LIBS arrow_shared)
 set(PLASMA_STATIC_LINK_LIBS arrow_static)
@@ -74,6 +111,8 @@ add_arrow_lib(plasma
               ${PLASMA_SRCS}
               OUTPUTS
               PLASMA_LIBRARIES
+              DEPENDENCIES
+              gen_plasma_fbs
               SHARED_LINK_FLAGS
               ${PLASMA_SHARED_LINK_FLAGS}
               SHARED_LINK_LIBS
@@ -115,8 +154,8 @@ else()
   # Fallback to shared libs in the case that static libraries are not build.
   target_link_libraries(plasma-store-server plasma_shared ${PLASMA_LINK_LIBS})
 endif()
-add_dependencies(plasma plasma-store-server)
 
+add_dependencies(plasma plasma-store-server)
 if(ARROW_RPATH_ORIGIN)
   if(APPLE)
     set(_lib_install_rpath "@loader_path")