mirror of
https://github.com/wassname/ray.git
synced 2026-07-27 11:26:41 +08:00
Make photon client into a C extension. (#8)
* Make photon client into a C extension. * Fix formatting. * Rename extension from PhotonClient to Photon. * Update common submodule. * Fix Makefile to compile with fPIC. * Update common submodule. * Compile C extension against common. * Fix formatting. * Remove unnecessary include. * Update common submodule and rename Photon -> PhotonClient. * Drop global interpretor lock during get_task.
This commit is contained in:
Robert Nishihara
committed by
Philipp Moritz
Philipp Moritz
parent
a048ad954a
commit
18934c3a45
+9
-3
@@ -44,11 +44,17 @@ matrix:
|
||||
- python test/test.py valgrind
|
||||
|
||||
install:
|
||||
- ./install-dependencies.sh
|
||||
- make
|
||||
|
||||
script:
|
||||
- cd common/lib/python
|
||||
- python setup.py install --user
|
||||
- cd ../../..
|
||||
- cd lib/python
|
||||
- python setup.py install --user
|
||||
- cd ../..
|
||||
- cd common
|
||||
- make test
|
||||
- cd ..
|
||||
- source setup-env.sh
|
||||
|
||||
script:
|
||||
- python test/test.py
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
CC = gcc
|
||||
CFLAGS = -g -Wall --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L -Icommon/thirdparty
|
||||
CFLAGS = -g -Wall --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L -Icommon/thirdparty -fPIC
|
||||
BUILD = build
|
||||
|
||||
all: $(BUILD)/photon_scheduler $(BUILD)/photon_client.so
|
||||
all: $(BUILD)/photon_scheduler $(BUILD)/photon_client.a
|
||||
|
||||
$(BUILD)/photon_client.so: photon_client.h photon_client.c common
|
||||
$(CC) $(CFLAGS) photon_client.c common/build/libcommon.a -fPIC -shared -o $(BUILD)/photon_client.so
|
||||
$(BUILD)/photon_client.a: photon_client.o
|
||||
ar rcs $(BUILD)/photon_client.a photon_client.o
|
||||
|
||||
$(BUILD)/photon_scheduler: photon.h photon_scheduler.c common
|
||||
$(CC) $(CFLAGS) -o $@ photon_scheduler.c common/build/libcommon.a common/thirdparty/hiredis/libhiredis.a -Icommon/thirdparty -Icommon/
|
||||
@@ -17,5 +17,6 @@ common: FORCE
|
||||
clean:
|
||||
cd common; make clean
|
||||
rm -r $(BUILD)/*
|
||||
rm *.o
|
||||
|
||||
FORCE:
|
||||
|
||||
+1
-1
Submodule common updated: 4204500d23...7be1a93d64
Executable
+21
@@ -0,0 +1,21 @@
|
||||
#!/usr/bin/env bash
|
||||
|
||||
ROOT_DIR=$(cd "$(dirname "${BASH_SOURCE:-$0}")"; pwd)
|
||||
|
||||
platform="unknown"
|
||||
unamestr="$(uname)"
|
||||
if [[ "$unamestr" == "Linux" ]]; then
|
||||
echo "Platform is linux."
|
||||
platform="linux"
|
||||
elif [[ "$unamestr" == "Darwin" ]]; then
|
||||
echo "Platform is macosx."
|
||||
platform="macosx"
|
||||
else
|
||||
echo "Unrecognized platform."
|
||||
exit 1
|
||||
fi
|
||||
|
||||
if [[ $platform == "linux" ]]; then
|
||||
sudo apt-get update
|
||||
sudo apt-get install -y git python-dev
|
||||
fi
|
||||
@@ -1,99 +0,0 @@
|
||||
import ctypes
|
||||
import os
|
||||
|
||||
photon_client_library_path = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../../build/photon_client.so")
|
||||
photon_client_library = ctypes.cdll.LoadLibrary(photon_client_library_path)
|
||||
photon_client_library.alloc_task_spec.restype = ctypes.c_void_p
|
||||
photon_client_library.photon_connect.restype = ctypes.c_void_p
|
||||
photon_client_library.photon_submit.restype = None
|
||||
photon_client_library.photon_get_task.restype = ctypes.c_void_p
|
||||
|
||||
ID = ctypes.c_ubyte * 20
|
||||
|
||||
buffer_from_read_write_memory = ctypes.pythonapi.PyBuffer_FromReadWriteMemory
|
||||
buffer_from_read_write_memory.argtypes = [ctypes.c_void_p, ctypes.c_int64]
|
||||
buffer_from_read_write_memory.restype = ctypes.py_object
|
||||
|
||||
buffer_from_memory = ctypes.pythonapi.PyBuffer_FromMemory
|
||||
buffer_from_memory.argtypes = [ctypes.c_void_p, ctypes.c_int64]
|
||||
buffer_from_memory.restype = ctypes.py_object
|
||||
|
||||
photon_client_library.task_function.restype = ctypes.c_void_p
|
||||
photon_client_library.task_num_args.restype = ctypes.c_int64
|
||||
photon_client_library.task_num_returns.restype = ctypes.c_int64
|
||||
photon_client_library.task_arg_type.restype = ctypes.c_int8
|
||||
photon_client_library.task_arg_id.restype = ctypes.c_void_p
|
||||
photon_client_library.task_arg_val.restype = ctypes.c_void_p
|
||||
photon_client_library.task_arg_length.restype = ctypes.c_void_p
|
||||
photon_client_library.task_return.restype = ctypes.c_void_p
|
||||
|
||||
|
||||
class TaskInfo(object):
|
||||
def __init__(self, function_id, args, return_ids):
|
||||
self.function_id = function_id
|
||||
self.args = args
|
||||
self.return_ids = return_ids
|
||||
|
||||
def extract_task(c_task):
|
||||
function_id = buffer_from_memory(photon_client_library.task_function(c_task), 20)[:]
|
||||
num_args = photon_client_library.task_num_args(c_task)
|
||||
num_returns = photon_client_library.task_num_returns(c_task)
|
||||
arg_vals_and_ids = []
|
||||
for i in range(num_args):
|
||||
arg_type = photon_client_library.task_arg_type(c_task, i)
|
||||
if arg_type == 0:
|
||||
arg_id = buffer_from_memory(photon_client_library.task_arg_id(c_task, i), 20)
|
||||
arg_vals_and_ids.append((arg_type, arg_id))
|
||||
elif arg_type == 1:
|
||||
arg_val = photon_client_library.task_arg_val(c_task, i)[:]
|
||||
arg_length = photon_client_library.task_arg_length(c_task, i)
|
||||
arg_value = buffer_from_memory(arg_val, arg_length)[:]
|
||||
arg_vals_and_ids.append((arg_type, arg_value))
|
||||
else:
|
||||
raise Exception("arg_type must be 0 or 1")
|
||||
return_ids = []
|
||||
for i in range(num_returns):
|
||||
ret_id = buffer_from_memory(photon_client_library.task_return(c_task, i), 20)
|
||||
return_ids.append(ret_id[:])
|
||||
return TaskInfo(function_id, arg_vals_and_ids, return_ids)
|
||||
|
||||
class UniqueID(ctypes.Structure):
|
||||
_fields_ = [("unique_id", ID)]
|
||||
|
||||
def make_id(string):
|
||||
if len(string) != 20:
|
||||
raise Exception("PlasmaIDs must be 20 characters long")
|
||||
unique_id = map(ord, string)
|
||||
return UniqueID(unique_id=ID(*unique_id))
|
||||
|
||||
class Task(object):
|
||||
def __init__(self, function_id, args, return_ids):
|
||||
function_id = make_id(function_id)
|
||||
self.task_spec = ctypes.c_void_p(photon_client_library.alloc_task_spec(function_id, len(args), 1, 0))
|
||||
for arg in args:
|
||||
photon_client_library.task_args_add_ref(self.task_spec, make_id(arg))
|
||||
|
||||
# Add return IDs. This may not be the appropriate place for this.
|
||||
num_returns = photon_client_library.task_num_returns(self.task_spec)
|
||||
for i in range(num_returns):
|
||||
ret_id = buffer_from_read_write_memory(photon_client_library.task_return(self.task_spec, i), 20)
|
||||
for j in range(20):
|
||||
ret_id[j] = return_ids[i][j]
|
||||
|
||||
def __del__(self):
|
||||
photon_client_library.free_task_spec(self.task_spec)
|
||||
|
||||
class PhotonClient(object):
|
||||
|
||||
def __init__(self, socket_name):
|
||||
self.photon_conn = ctypes.c_void_p(photon_client_library.photon_connect(socket_name))
|
||||
|
||||
def submit(self, function_id, args, return_ids):
|
||||
task = Task(function_id, args, return_ids)
|
||||
photon_client_library.photon_submit(self.photon_conn, task.task_spec)
|
||||
|
||||
def get_task(self):
|
||||
c_task = ctypes.c_void_p(photon_client_library.photon_get_task(self.photon_conn))
|
||||
task = c_task # TODO Extract the actual task. EXTRACT...(c_task)
|
||||
# photon_client_library.free_task_spec(c_task)
|
||||
return extract_task(task)
|
||||
@@ -0,0 +1,140 @@
|
||||
#include <Python.h>
|
||||
|
||||
#include "common_extension.h"
|
||||
#include "photon_client.h"
|
||||
#include "task.h"
|
||||
|
||||
PyObject *PhotonError;
|
||||
|
||||
// clang-format off
|
||||
typedef struct {
|
||||
PyObject_HEAD
|
||||
photon_conn *photon_connection;
|
||||
} PyPhotonClient;
|
||||
// clang-format on
|
||||
|
||||
static int PyPhotonClient_init(PyPhotonClient *self, PyObject *args,
|
||||
PyObject *kwds) {
|
||||
char *socket_name;
|
||||
if (!PyArg_ParseTuple(args, "s", &socket_name)) {
|
||||
return -1;
|
||||
}
|
||||
self->photon_connection = photon_connect(socket_name);
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void PyPhotonClient_dealloc(PyPhotonClient *self) {
|
||||
free(((PyPhotonClient *)self)->photon_connection);
|
||||
Py_TYPE(self)->tp_free((PyObject *)self);
|
||||
}
|
||||
|
||||
static PyObject *PyPhotonClient_submit(PyObject *self, PyObject *args) {
|
||||
PyObject *py_task;
|
||||
if (!PyArg_ParseTuple(args, "O", &py_task)) {
|
||||
return NULL;
|
||||
}
|
||||
photon_submit(((PyPhotonClient *)self)->photon_connection,
|
||||
((PyTask *)py_task)->spec);
|
||||
Py_RETURN_NONE;
|
||||
}
|
||||
|
||||
// clang-format off
|
||||
static PyObject *PyPhotonClient_get_task(PyObject *self) {
|
||||
task_spec *task_spec;
|
||||
/* Drop the global interpreter lock while we get a task because
|
||||
* photon_get_task may block for a long time. */
|
||||
Py_BEGIN_ALLOW_THREADS
|
||||
task_spec = photon_get_task(((PyPhotonClient *)self)->photon_connection);
|
||||
Py_END_ALLOW_THREADS
|
||||
return PyTask_make(task_spec);
|
||||
}
|
||||
// clang-format on
|
||||
|
||||
static PyMethodDef PyPhotonClient_methods[] = {
|
||||
{"submit", (PyCFunction)PyPhotonClient_submit, METH_VARARGS,
|
||||
"Submit a task to the local scheduler."},
|
||||
{"get_task", (PyCFunction)PyPhotonClient_get_task, METH_NOARGS,
|
||||
"Get a task from the local scheduler."},
|
||||
{NULL} /* Sentinel */
|
||||
};
|
||||
|
||||
static PyTypeObject PyPhotonClientType = {
|
||||
PyObject_HEAD_INIT(NULL) 0, /* ob_size */
|
||||
"photon.PhotonClient", /* tp_name */
|
||||
sizeof(PyPhotonClient), /* tp_basicsize */
|
||||
0, /* tp_itemsize */
|
||||
(destructor)PyPhotonClient_dealloc, /* tp_dealloc */
|
||||
0, /* tp_print */
|
||||
0, /* tp_getattr */
|
||||
0, /* tp_setattr */
|
||||
0, /* tp_compare */
|
||||
0, /* tp_repr */
|
||||
0, /* tp_as_number */
|
||||
0, /* tp_as_sequence */
|
||||
0, /* tp_as_mapping */
|
||||
0, /* tp_hash */
|
||||
0, /* tp_call */
|
||||
0, /* tp_str */
|
||||
0, /* tp_getattro */
|
||||
0, /* tp_setattro */
|
||||
0, /* tp_as_buffer */
|
||||
Py_TPFLAGS_DEFAULT, /* tp_flags */
|
||||
"PhotonClient object", /* tp_doc */
|
||||
0, /* tp_traverse */
|
||||
0, /* tp_clear */
|
||||
0, /* tp_richcompare */
|
||||
0, /* tp_weaklistoffset */
|
||||
0, /* tp_iter */
|
||||
0, /* tp_iternext */
|
||||
PyPhotonClient_methods, /* tp_methods */
|
||||
0, /* tp_members */
|
||||
0, /* tp_getset */
|
||||
0, /* tp_base */
|
||||
0, /* tp_dict */
|
||||
0, /* tp_descr_get */
|
||||
0, /* tp_descr_set */
|
||||
0, /* tp_dictoffset */
|
||||
(initproc)PyPhotonClient_init, /* tp_init */
|
||||
0, /* tp_alloc */
|
||||
PyType_GenericNew, /* tp_new */
|
||||
};
|
||||
|
||||
static PyMethodDef photon_methods[] = {
|
||||
{"check_simple_value", check_simple_value, METH_VARARGS,
|
||||
"Should the object be passed by value?"},
|
||||
{NULL} /* Sentinel */
|
||||
};
|
||||
|
||||
#ifndef PyMODINIT_FUNC /* declarations for DLL import/export */
|
||||
#define PyMODINIT_FUNC void
|
||||
#endif
|
||||
|
||||
PyMODINIT_FUNC initphoton(void) {
|
||||
PyObject *m;
|
||||
|
||||
if (PyType_Ready(&PyTaskType) < 0)
|
||||
return;
|
||||
|
||||
if (PyType_Ready(&PyObjectIDType) < 0)
|
||||
return;
|
||||
|
||||
if (PyType_Ready(&PyPhotonClientType) < 0)
|
||||
return;
|
||||
|
||||
m = Py_InitModule3("photon", photon_methods,
|
||||
"A module for the local scheduler.");
|
||||
|
||||
Py_INCREF(&PyTaskType);
|
||||
PyModule_AddObject(m, "Task", (PyObject *)&PyTaskType);
|
||||
|
||||
Py_INCREF(&PyObjectIDType);
|
||||
PyModule_AddObject(m, "ObjectID", (PyObject *)&PyObjectIDType);
|
||||
|
||||
Py_INCREF(&PyPhotonClientType);
|
||||
PyModule_AddObject(m, "PhotonClient", (PyObject *)&PyPhotonClientType);
|
||||
|
||||
char photon_error[] = "photon.error";
|
||||
PhotonError = PyErr_NewException(photon_error, NULL, NULL);
|
||||
Py_INCREF(PhotonError);
|
||||
PyModule_AddObject(m, "photon_error", PhotonError);
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
from setuptools import setup, find_packages, Extension
|
||||
|
||||
photon_module = Extension("photon",
|
||||
sources=["photon_extension.c", "../../common/lib/python/common_extension.c"],
|
||||
include_dirs=["../../", "../../common/",
|
||||
"../../common/thirdparty/",
|
||||
"../../common/lib/python"],
|
||||
extra_objects=["../../build/photon_client.a", "../../common/build/libcommon.a"],
|
||||
extra_compile_args=["--std=c99", "-Werror"])
|
||||
|
||||
setup(name="Photon",
|
||||
version="0.1",
|
||||
description="Photon library for Ray",
|
||||
ext_modules=[photon_module])
|
||||
+48
-7
@@ -45,13 +45,54 @@ class TestPhotonClient(unittest.TestCase):
|
||||
self.p2.kill()
|
||||
|
||||
|
||||
def test_create(self):
|
||||
l = [20 * "a", 20 * "b", 20 * "c"]
|
||||
r = [20 * "e", 20 * "f"]
|
||||
# Submit a task.
|
||||
self.photon_client.submit(20 * "d", l, r)
|
||||
# Get the task.
|
||||
task = self.photon_client.get_task()
|
||||
def test_submit_and_get_task(self):
|
||||
# TODO(rkn): This should be a FunctionID.
|
||||
function_id = photon.ObjectID(20 * "a")
|
||||
object_ids = [photon.ObjectID(20 * chr(i)) for i in range(256)]
|
||||
args_list = [
|
||||
[],
|
||||
1 * [1],
|
||||
10 * [1],
|
||||
100 * [1],
|
||||
1000 * [1],
|
||||
1 * ["a"],
|
||||
10 * ["a"],
|
||||
100 * ["a"],
|
||||
1000 * ["a"],
|
||||
[1, 1.3, 2L, 1L << 100, "hi", u"hi", [1, 2]],
|
||||
object_ids[:1],
|
||||
object_ids[:2],
|
||||
object_ids[:3],
|
||||
object_ids[:4],
|
||||
object_ids[:5],
|
||||
object_ids[:10],
|
||||
object_ids[:100],
|
||||
object_ids[:256],
|
||||
[1, object_ids[0]],
|
||||
[object_ids[0], "a"],
|
||||
[1, object_ids[0], "a"],
|
||||
[object_ids[0], 1, object_ids[1], "a"],
|
||||
object_ids[:3] + [1, "hi", 2.3] + object_ids[:5],
|
||||
object_ids + 100 * ["a"] + object_ids
|
||||
]
|
||||
|
||||
for args in args_list:
|
||||
for num_return_vals in [0, 1, 2, 3, 5, 10, 100]:
|
||||
task = photon.Task(function_id, args, num_return_vals)
|
||||
# Submit a task.
|
||||
self.photon_client.submit(task)
|
||||
# Get the task.
|
||||
new_task = self.photon_client.get_task()
|
||||
self.assertEqual(task.function_id().id(), new_task.function_id().id())
|
||||
retrieved_args = new_task.arguments()
|
||||
returns = new_task.returns()
|
||||
self.assertEqual(len(args), len(retrieved_args))
|
||||
self.assertEqual(num_return_vals, len(returns))
|
||||
for i in range(len(retrieved_args)):
|
||||
if isinstance(args[i], photon.ObjectID):
|
||||
self.assertEqual(args[i].id(), retrieved_args[i].id())
|
||||
else:
|
||||
self.assertEqual(args[i], retrieved_args[i])
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) > 1:
|
||||
|
||||
Reference in New Issue
Block a user