Skip to content

Commit a94f471

Browse files
pcmoritzwesm
authored andcommitted
ARROW-1149: [Plasma] Create Cython client library for Plasma
This PR introduces a Cython API to Plasma, a FindPlasma.cmake to make it easier to integrate Plasma with CMake projects and sets up packaging with pyarrow. Author: Philipp Moritz <pcmoritz@gmail.com> Author: Robert Nishihara <robertnishihara@gmail.com> Closes apache#797 from pcmoritz/plasma-cython and squashes the following commits: d8319fc [Philipp Moritz] get for of PlasmaClient.connect d14ab87 [Philipp Moritz] get rid of MutableBuffer 08f24a5 [Philipp Moritz] fix typos and move FixedSizeBufferOutputStream e33443d [Philipp Moritz] fix setup.py develop for plasma 5f7b779 [Philipp Moritz] changes needed to make Ray work with Plasma in Arrow b9e2dee [Philipp Moritz] fix windows build 3e4a84d [Philipp Moritz] fix segfault 0bea267 [Philipp Moritz] debug 23fe5f5 [Philipp Moritz] make plasma store binary part of the pyarrow package for tests b863d13 [Philipp Moritz] fix 997de1e [Philipp Moritz] fix 47dc739 [Philipp Moritz] fixes 47033e7 [Philipp Moritz] switch to pytest ed84c53 [Philipp Moritz] partial fixes 9bc5c15 [Philipp Moritz] implement wait and fetch for the client 45f338f [Philipp Moritz] test plasma on macOS 8b53618 [Philipp Moritz] fix 54f595e [Philipp Moritz] try fixing python 2 tests 2c6d652 [Philipp Moritz] convert docs to numpy format 3270628 [Philipp Moritz] try to get documentation up 44d1a55 [Philipp Moritz] cleanups and release GIL a9f6502 [Philipp Moritz] more fixes 1ff88e7 [Philipp Moritz] fix travix ci 348f9bf [Philipp Moritz] fixes 4ae1a27 [Philipp Moritz] fix fd80203 [Philipp Moritz] Plasma Python extension packaging: It compiles! 3b69973 [Robert Nishihara] Fixed minor python linting. c9f6bcf [Robert Nishihara] Fix indentation and line lengths in plasma.pyx. 67b0951 [Robert Nishihara] Fix long lines in plasma/test/test.py. e26527c [Robert Nishihara] Convert plasma test.py from 2 space indentation to 4 space indentation. acc71d2 [Philipp Moritz] add round trip test for dataframes 2b7f949 [Philipp Moritz] implement mutable arrow python buffers c06f1b5 [Philipp Moritz] fix test 1d7928f [Philipp Moritz] add arrow roundtrip test 6371e2e [Philipp Moritz] fix tests 3021d59 [Philipp Moritz] make ObjectID pickleable dd5a7d8 [Philipp Moritz] fix tests 777e9c7 [Philipp Moritz] introduce plasma namespace a4a9628 [Philipp Moritz] fix c++ tests 924888b [Philipp Moritz] update f970df3 [Philipp Moritz] reduce logging 2ff2480 [Philipp Moritz] workaround for python visibility d4934a9 [Philipp Moritz] update cba92c1 [Philipp Moritz] setup.py for plasma 066d0ea [Philipp Moritz] test 1aea320 [Philipp Moritz] run plasma tests 3c4de52 [Philipp Moritz] use cmake to build the cython extension bf39297 [Philipp Moritz] build and install pyarrow for plasma tests 5bf722a [Philipp Moritz] fix plasma path 1c5434c [Philipp Moritz] fix formatting 187cc24 [Philipp Moritz] add travis tests c3d462d [Philipp Moritz] remove Python C extension d9261b4 [Philipp Moritz] add documentation and license db2d09a [Philipp Moritz] get all python tests in place 78d08ac [Philipp Moritz] make eviction work in Cython 18e0ac4 [Philipp Moritz] get tests bc681ca [Philipp Moritz] port some python tests f8e05f2 [Philipp Moritz] implement plasma.get in the cython client d590c8a [Philipp Moritz] update 5178ee7 [Philipp Moritz] update 9044a01 [Philipp Moritz] initial plasma cython client commit
1 parent 05f7058 commit a94f471

38 files changed

Lines changed: 1855 additions & 631 deletions

‎.travis.yml‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,6 +120,27 @@ matrix:
120120
- $TRAVIS_BUILD_DIR/ci/travis_before_script_c_glib.sh
121121
script:
122122
- $TRAVIS_BUILD_DIR/ci/travis_script_c_glib.sh
123+
- compiler: gcc
124+
language: cpp
125+
os: linux
126+
group: deprecated
127+
before_script:
128+
- export CC="gcc-4.9"
129+
- export CXX="g++-4.9"
130+
- $TRAVIS_BUILD_DIR/ci/travis_before_script_cpp.sh
131+
script:
132+
- $TRAVIS_BUILD_DIR/ci/travis_script_cpp.sh
133+
- $TRAVIS_BUILD_DIR/ci/travis_script_plasma.sh
134+
- compiler: clang
135+
osx_image: xcode6.4
136+
os: osx
137+
cache:
138+
addons:
139+
before_script:
140+
- $TRAVIS_BUILD_DIR/ci/travis_before_script_cpp.sh
141+
script:
142+
- $TRAVIS_BUILD_DIR/ci/travis_script_cpp.sh
143+
- $TRAVIS_BUILD_DIR/ci/travis_script_plasma.sh
123144

124145
before_install:
125146
- ulimit -c unlimited -S

‎ci/travis_script_manylinux.sh‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,4 +18,4 @@ set -ex
1818
pushd python/manylinux1
1919
git clone ../../ arrow
2020
docker build -t arrow-base-x86_64 -f Dockerfile-x86_64 .
21-
docker run --rm -e PYARROW_PARALLEL=3 -v $PWD:/io arrow-base-x86_64 /io/build_arrow.sh
21+
docker run --shm-size=2g --rm -e PYARROW_PARALLEL=3 -v $PWD:/io arrow-base-x86_64 /io/build_arrow.sh

‎ci/travis_script_plasma.sh‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
#!/usr/bin/env bash
2+
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License. See accompanying LICENSE file.
14+
15+
set -e
16+
17+
source $TRAVIS_BUILD_DIR/ci/travis_env_common.sh
18+
19+
export ARROW_HOME=$ARROW_CPP_INSTALL
20+
export PYARROW_WITH_PLASMA=1
21+
22+
pushd $ARROW_PYTHON_DIR
23+
24+
function build_arrow_libraries() {
25+
CPP_BUILD_DIR=$1
26+
CPP_DIR=$TRAVIS_BUILD_DIR/cpp
27+
28+
mkdir $CPP_BUILD_DIR
29+
pushd $CPP_BUILD_DIR
30+
31+
cmake -DARROW_BUILD_TESTS=off \
32+
-DARROW_PYTHON=on \
33+
-DARROW_PLASMA=on \
34+
-DCMAKE_INSTALL_PREFIX=$2 \
35+
$CPP_DIR
36+
37+
make -j4
38+
make install
39+
40+
popd
41+
}
42+
43+
python_version_tests() {
44+
PYTHON_VERSION=$1
45+
CONDA_ENV_DIR=$TRAVIS_BUILD_DIR/pyarrow-test-$PYTHON_VERSION
46+
47+
export ARROW_HOME=$TRAVIS_BUILD_DIR/arrow-install-$PYTHON_VERSION
48+
export LD_LIBRARY_PATH=$ARROW_HOME/lib:$PARQUET_HOME/lib
49+
50+
conda create -y -q -p $CONDA_ENV_DIR python=$PYTHON_VERSION cmake curl
51+
source activate $CONDA_ENV_DIR
52+
53+
python --version
54+
which python
55+
56+
# faster builds, please
57+
conda install -y -q nomkl
58+
59+
# Expensive dependencies install from Continuum package repo
60+
conda install -y -q pip numpy pandas cython
61+
62+
# Build C++ libraries
63+
build_arrow_libraries arrow-build-$PYTHON_VERSION $ARROW_HOME
64+
65+
# Other stuff pip install
66+
pip install -r requirements.txt
67+
68+
python setup.py build_ext --inplace
69+
70+
python -m pytest -vv -r sxX pyarrow
71+
72+
# Build documentation once
73+
if [[ "$PYTHON_VERSION" == "3.6" ]]
74+
then
75+
conda install -y -q --file=doc/requirements.txt
76+
python setup.py build_sphinx -s doc/source
77+
fi
78+
79+
# Build and install pyarrow
80+
pushd $TRAVIS_BUILD_DIR/python
81+
python setup.py install
82+
popd
83+
84+
# Run Plasma tests
85+
pushd $TRAVIS_BUILD_DIR/python
86+
python -m pytest pyarrow/tests/test_plasma.py
87+
if [ $TRAVIS_OS_NAME == "linux" ]; then
88+
PLASMA_VALGRIND=1 python -m pytest pyarrow/tests/test_plasma.py
89+
fi
90+
popd
91+
}
92+
93+
# run tests for python 2.7 and 3.6
94+
python_version_tests 2.7
95+
python_version_tests 3.6
96+
97+
popd

‎ci/travis_script_python.sh‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ set -e
1717
source $TRAVIS_BUILD_DIR/ci/travis_env_common.sh
1818

1919
export ARROW_HOME=$ARROW_CPP_INSTALL
20+
export PYARROW_WITH_PLASMA=1
2021

2122
pushd $ARROW_PYTHON_DIR
2223
export PARQUET_HOME=$TRAVIS_BUILD_DIR/parquet-env
@@ -71,9 +72,8 @@ function build_arrow_libraries() {
7172
pushd $CPP_BUILD_DIR
7273

7374
cmake -DARROW_BUILD_TESTS=off \
74-
-DARROW_PYTHON=on \
75-
-DPLASMA_PYTHON=on \
7675
-DARROW_PLASMA=on \
76+
-DARROW_PYTHON=on \
7777
-DCMAKE_INSTALL_PREFIX=$2 \
7878
$CPP_DIR
7979

‎cpp/src/arrow/util/logging.h‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -113,8 +113,10 @@ class CerrLog {
113113

114114
template <class T>
115115
CerrLog& operator<<(const T& t) {
116-
has_logged_ = true;
117-
std::cerr << t;
116+
if (severity_ != ARROW_DEBUG) {
117+
has_logged_ = true;
118+
std::cerr << t;
119+
}
118120
return *this;
119121
}
120122

‎cpp/src/plasma/CMakeLists.txt‎

Lines changed: 31 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -19,16 +19,13 @@ cmake_minimum_required(VERSION 2.8)
1919

2020
project(plasma)
2121

22+
set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${CMAKE_SOURCE_DIR}/../python/cmake_modules")
23+
2224
find_package(PythonLibsNew REQUIRED)
2325
find_package(Threads)
2426

25-
option(PLASMA_PYTHON
26-
"Build the Plasma Python extensions"
27-
OFF)
28-
29-
if(APPLE)
30-
SET(CMAKE_SHARED_LIBRARY_SUFFIX ".so")
31-
endif(APPLE)
27+
set(PLASMA_SO_VERSION "0")
28+
set(PLASMA_ABI_VERSION "${PLASMA_SO_VERSION}.0.0")
3229

3330
include_directories(SYSTEM ${PYTHON_INCLUDE_DIRS})
3431
include_directories("${FLATBUFFERS_INCLUDE_DIR}" "${CMAKE_CURRENT_LIST_DIR}/" "${CMAKE_CURRENT_LIST_DIR}/thirdparty/" "${CMAKE_CURRENT_LIST_DIR}/../")
@@ -40,7 +37,7 @@ set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wno-conversion")
4037
# Compile flatbuffers
4138

4239
set(PLASMA_FBS_SRC "${CMAKE_CURRENT_LIST_DIR}/format/plasma.fbs" "${CMAKE_CURRENT_LIST_DIR}/format/common.fbs")
43-
set(OUTPUT_DIR ${CMAKE_CURRENT_LIST_DIR}/format/)
40+
set(OUTPUT_DIR ${CMAKE_CURRENT_LIST_DIR}/)
4441

4542
set(PLASMA_FBS_OUTPUT_FILES
4643
"${OUTPUT_DIR}/common_generated.h"
@@ -69,8 +66,6 @@ endif()
6966

7067
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fPIC")
7168

72-
set_source_files_properties(extension.cc PROPERTIES COMPILE_FLAGS -Wno-strict-aliasing)
73-
7469
set(PLASMA_SRCS
7570
client.cc
7671
common.cc
@@ -97,17 +92,33 @@ set_source_files_properties(malloc.cc PROPERTIES COMPILE_FLAGS "-Wno-error -O3")
9792
add_executable(plasma_store store.cc)
9893
target_link_libraries(plasma_store plasma_static)
9994

95+
# Headers: top level
96+
install(FILES
97+
common.h
98+
common_generated.h
99+
client.h
100+
events.h
101+
plasma.h
102+
plasma_generated.h
103+
protocol.h
104+
DESTINATION "${CMAKE_INSTALL_INCLUDEDIR}/plasma")
105+
106+
# Plasma store
107+
install(TARGETS plasma_store DESTINATION ${CMAKE_INSTALL_BINDIR})
108+
109+
# pkg-config support
110+
configure_file(plasma.pc.in
111+
"${CMAKE_CURRENT_BINARY_DIR}/plasma.pc"
112+
@ONLY)
113+
install(
114+
FILES "${CMAKE_CURRENT_BINARY_DIR}/plasma.pc"
115+
DESTINATION "${CMAKE_INSTALL_LIBDIR}/pkgconfig/")
116+
117+
#######################################
118+
# Unit tests
119+
#######################################
120+
100121
ADD_ARROW_TEST(test/serialization_tests)
101122
ARROW_TEST_LINK_LIBRARIES(test/serialization_tests plasma_static)
102123
ADD_ARROW_TEST(test/client_tests)
103124
ARROW_TEST_LINK_LIBRARIES(test/client_tests plasma_static)
104-
105-
if(PLASMA_PYTHON)
106-
add_library(plasma_extension SHARED extension.cc)
107-
108-
if(APPLE)
109-
target_link_libraries(plasma_extension plasma_static "-undefined dynamic_lookup")
110-
else(APPLE)
111-
target_link_libraries(plasma_extension plasma_static -Wl,--whole-archive ${FLATBUFFERS_STATIC_LIB} -Wl,--no-whole-archive)
112-
endif(APPLE)
113-
endif()

‎cpp/src/plasma/client.cc‎

Lines changed: 68 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -51,11 +51,31 @@
5151

5252
#define XXH64_DEFAULT_SEED 0
5353

54+
namespace plasma {
55+
5456
// Number of threads used for memcopy and hash computations.
5557
constexpr int64_t kThreadPoolSize = 8;
5658
constexpr int64_t kBytesInMB = 1 << 20;
5759
static std::vector<std::thread> threadpool_(kThreadPoolSize);
5860

61+
struct ObjectInUseEntry {
62+
/// A count of the number of times this client has called PlasmaClient::Create
63+
/// or
64+
/// PlasmaClient::Get on this object ID minus the number of calls to
65+
/// PlasmaClient::Release.
66+
/// When this count reaches zero, we remove the entry from the ObjectsInUse
67+
/// and decrement a count in the relevant ClientMmapTableEntry.
68+
int count;
69+
/// Cached information to read the object.
70+
PlasmaObject object;
71+
/// A flag representing whether the object has been sealed.
72+
bool is_sealed;
73+
};
74+
75+
PlasmaClient::PlasmaClient() {}
76+
77+
PlasmaClient::~PlasmaClient() {}
78+
5979
// If the file descriptor fd has been mmapped in this client process before,
6080
// return the pointer that was returned by mmap, otherwise mmap it and store the
6181
// pointer in a hash table.
@@ -300,6 +320,10 @@ Status PlasmaClient::PerformRelease(const ObjectID& object_id) {
300320
}
301321

302322
Status PlasmaClient::Release(const ObjectID& object_id) {
323+
// If the client is already disconnected, ignore release requests.
324+
if (store_conn_ < 0) {
325+
return Status::OK();
326+
}
303327
// Add the new object to the release history.
304328
release_history_.push_front(object_id);
305329
// If there are too many bytes in use by the client or if there are too many
@@ -386,22 +410,6 @@ static uint64_t compute_object_hash(const ObjectBuffer& obj_buffer) {
386410
return XXH64_digest(&hash_state);
387411
}
388412

389-
bool plasma_compute_object_hash(
390-
PlasmaClient* conn, ObjectID object_id, unsigned char* digest) {
391-
// Get the plasma object data. We pass in a timeout of 0 to indicate that
392-
// the operation should timeout immediately.
393-
ObjectBuffer object_buffer;
394-
ARROW_CHECK_OK(conn->Get(&object_id, 1, 0, &object_buffer));
395-
// If the object was not retrieved, return false.
396-
if (object_buffer.data_size == -1) { return false; }
397-
// Compute the hash.
398-
uint64_t hash = compute_object_hash(object_buffer);
399-
memcpy(digest, &hash, sizeof(hash));
400-
// Release the plasma object.
401-
ARROW_CHECK_OK(conn->Release(object_id));
402-
return true;
403-
}
404-
405413
Status PlasmaClient::Seal(const ObjectID& object_id) {
406414
// Make sure this client has a reference to the object before sending the
407415
// request to Plasma.
@@ -413,7 +421,7 @@ Status PlasmaClient::Seal(const ObjectID& object_id) {
413421
object_entry->second->is_sealed = true;
414422
/// Send the seal request to Plasma.
415423
static unsigned char digest[kDigestSize];
416-
ARROW_CHECK(plasma_compute_object_hash(this, object_id, &digest[0]));
424+
RETURN_NOT_OK(Hash(object_id, &digest[0]));
417425
RETURN_NOT_OK(SendSealRequest(store_conn_, object_id, &digest[0]));
418426
// We call PlasmaClient::Release to decrement the number of instances of this
419427
// object
@@ -439,6 +447,22 @@ Status PlasmaClient::Evict(int64_t num_bytes, int64_t& num_bytes_evicted) {
439447
return ReadEvictReply(buffer.data(), num_bytes_evicted);
440448
}
441449

450+
Status PlasmaClient::Hash(const ObjectID& object_id, uint8_t* digest) {
451+
// Get the plasma object data. We pass in a timeout of 0 to indicate that
452+
// the operation should timeout immediately.
453+
ObjectBuffer object_buffer;
454+
RETURN_NOT_OK(Get(&object_id, 1, 0, &object_buffer));
455+
// If the object was not retrieved, return false.
456+
if (object_buffer.data_size == -1) {
457+
return Status::PlasmaObjectNonexistent("Object not found");
458+
}
459+
// Compute the hash.
460+
uint64_t hash = compute_object_hash(object_buffer);
461+
memcpy(digest, &hash, sizeof(hash));
462+
// Release the plasma object.
463+
return Release(object_id);
464+
}
465+
442466
Status PlasmaClient::Subscribe(int* fd) {
443467
int sock[2];
444468
// Create a non-blocking socket pair. This will only be used to send
@@ -459,6 +483,26 @@ Status PlasmaClient::Subscribe(int* fd) {
459483
return Status::OK();
460484
}
461485

486+
Status PlasmaClient::GetNotification(
487+
int fd, ObjectID* object_id, int64_t* data_size, int64_t* metadata_size) {
488+
uint8_t* notification = read_message_async(fd);
489+
if (notification == NULL) {
490+
return Status::IOError("Failed to read object notification from Plasma socket");
491+
}
492+
auto object_info = flatbuffers::GetRoot<ObjectInfo>(notification);
493+
ARROW_CHECK(object_info->object_id()->size() == sizeof(ObjectID));
494+
memcpy(object_id, object_info->object_id()->data(), sizeof(ObjectID));
495+
if (object_info->is_deletion()) {
496+
*data_size = -1;
497+
*metadata_size = -1;
498+
} else {
499+
*data_size = object_info->data_size();
500+
*metadata_size = object_info->metadata_size();
501+
}
502+
delete[] notification;
503+
return Status::OK();
504+
}
505+
462506
Status PlasmaClient::Connect(const std::string& store_socket_name,
463507
const std::string& manager_socket_name, int release_delay) {
464508
store_conn_ = connect_ipc_sock_retry(store_socket_name, -1, -1);
@@ -485,7 +529,11 @@ Status PlasmaClient::Disconnect() {
485529
// Close the connections to Plasma. The Plasma store will release the objects
486530
// that were in use by us when handling the SIGPIPE.
487531
close(store_conn_);
488-
if (manager_conn_ >= 0) { close(manager_conn_); }
532+
store_conn_ = -1;
533+
if (manager_conn_ >= 0) {
534+
close(manager_conn_);
535+
manager_conn_ = -1;
536+
}
489537
return Status::OK();
490538
}
491539

@@ -555,3 +603,5 @@ Status PlasmaClient::Wait(int64_t num_object_requests, ObjectRequest* object_req
555603
}
556604
return Status::OK();
557605
}
606+
607+
} // namespace plasma

0 commit comments

Comments
 (0)