#include <Python.h>
#include "bytesobject.h"
#include "node.h"

// Don't use the deprecated Numpy functions.
#define NPY_NO_DEPRECATED_API NPY_1_7_API_VERSION

#include <numpy/arrayobject.h>

#include "common.h"
#include "common_extension.h"
#include "common_protocol.h"
#include "ray/raylet/task_spec.h"
#include "task.h"

#include <string>

#if PY_MAJOR_VERSION >= 3
#define PyInt_Check PyLong_Check
#endif

PyObject *CommonError;

/* Initialize pickle module. */

PyObject *pickle_module = NULL;
PyObject *pickle_loads = NULL;
PyObject *pickle_dumps = NULL;
PyObject *pickle_protocol = NULL;

int init_numpy_module(void) {
  import_array1(-1);
  return 0;
}

void init_pickle_module(void) {
#if PY_MAJOR_VERSION >= 3
  pickle_module = PyImport_ImportModule("pickle");
#else
  pickle_module = PyImport_ImportModuleNoBlock("cPickle");
#endif
  RAY_CHECK(pickle_module != NULL);
  RAY_CHECK(PyObject_HasAttrString(pickle_module, "loads"));
  RAY_CHECK(PyObject_HasAttrString(pickle_module, "dumps"));
  RAY_CHECK(PyObject_HasAttrString(pickle_module, "HIGHEST_PROTOCOL"));
  pickle_loads = PyUnicode_FromString("loads");
  pickle_dumps = PyUnicode_FromString("dumps");
  pickle_protocol = PyObject_GetAttrString(pickle_module, "HIGHEST_PROTOCOL");
  RAY_CHECK(pickle_protocol != NULL);
}

TaskBuilder *g_task_builder = NULL;

/* Define the PyObjectID class. */

int PyStringToUniqueID(PyObject *object, ObjectID *object_id) {
  if (PyBytes_Check(object)) {
    std::memcpy(object_id->mutable_data(), PyBytes_AsString(object),
                sizeof(*object_id));
    return 1;
  } else {
    PyErr_SetString(PyExc_TypeError, "must be a 20 character string");
    return 0;
  }
}

int PyObjectToUniqueID(PyObject *object, ObjectID *objectid) {
  if (PyObject_IsInstance(object, (PyObject *) &PyObjectIDType)) {
    *objectid = ((PyObjectID *) object)->object_id;
    return 1;
  } else {
    PyErr_SetString(PyExc_TypeError, "must be an ObjectID");
    return 0;
  }
}

bool use_raylet(PyTask *task) {
  return task->spec == nullptr;
}

static int PyObjectID_init(PyObjectID *self, PyObject *args, PyObject *kwds) {
  const char *data;
  int size;
  if (!PyArg_ParseTuple(args, "s#", &data, &size)) {
    return -1;
  }
  if (size != sizeof(ObjectID)) {
    PyErr_SetString(CommonError,
                    "ObjectID: object id string needs to have length 20");
    return -1;
  }
  std::memcpy(self->object_id.mutable_data(), data, sizeof(self->object_id));
  return 0;
}

/* Create a PyObjectID from C. */
PyObject *PyObjectID_make(ObjectID object_id) {
  PyObjectID *result = PyObject_New(PyObjectID, &PyObjectIDType);
  result = (PyObjectID *) PyObject_Init((PyObject *) result, &PyObjectIDType);
  result->object_id = object_id;
  return (PyObject *) result;
}

/**
 * Convert a string to a Ray task specification Python object.
 *
 * This is called from Python like
 *
 * task = local_scheduler.task_from_string("...")
 *
 * @param task_string String representation of the task specification.
 * @return Python task specification object.
 */
PyObject *PyTask_from_string(PyObject *self, PyObject *args) {
  const char *data;
  int size;
  if (!PyArg_ParseTuple(args, "s#", &data, &size)) {
    return NULL;
  }
  PyTask *result = PyObject_New(PyTask, &PyTaskType);
  result = (PyTask *) PyObject_Init((PyObject *) result, &PyTaskType);
  result->size = size;
  result->spec = TaskSpec_copy((TaskSpec *) data, size);
  /* The created task does not include any execution dependencies. */
  result->execution_dependencies = new std::vector<ObjectID>();
  /* TODO(pcm): Use flatbuffers validation here. */
  return (PyObject *) result;
}

/**
 * Convert a Ray task specification Python object to a string.
 *
 * This is called from Python like
 *
 * s = local_scheduler.task_to_string(task)
 *
 * @param task Ray task specification Python object.
 * @return String representing the task specification.
 */
PyObject *PyTask_to_string(PyObject *self, PyObject *args) {
  PyObject *arg;
  if (!PyArg_ParseTuple(args, "O", &arg)) {
    return NULL;
  }
  PyTask *task = (PyTask *) arg;
  return PyBytes_FromStringAndSize((char *) task->spec, task->size);
}

static PyObject *PyObjectID_id(PyObject *self) {
  PyObjectID *s = (PyObjectID *) self;
  return PyBytes_FromStringAndSize((const char *) s->object_id.data(),
                                   sizeof(s->object_id));
}

static PyObject *PyObjectID_hex(PyObject *self) {
  PyObjectID *s = (PyObjectID *) self;
  std::string hex_id = s->object_id.hex();
  PyObject *result = PyUnicode_FromString(hex_id.c_str());
  return result;
}

static PyObject *PyObjectID_richcompare(PyObjectID *self,
                                        PyObject *other,
                                        int op) {
  PyObject *result = NULL;
  if (Py_TYPE(self)->tp_richcompare != Py_TYPE(other)->tp_richcompare) {
    result = Py_NotImplemented;
  } else {
    PyObjectID *other_id = (PyObjectID *) other;
    switch (op) {
    case Py_LT:
      result = Py_NotImplemented;
      break;
    case Py_LE:
      result = Py_NotImplemented;
      break;
    case Py_EQ:
      result = self->object_id == other_id->object_id ? Py_True : Py_False;
      break;
    case Py_NE:
      result = !(self->object_id == other_id->object_id) ? Py_True : Py_False;
      break;
    case Py_GT:
      result = Py_NotImplemented;
      break;
    case Py_GE:
      result = Py_NotImplemented;
      break;
    }
  }
  Py_XINCREF(result);
  return result;
}

static PyObject *PyObjectID_redis_shard_hash(PyObjectID *self) {
  /* NOTE: The hash function used here must match the one in get_redis_context
   * in src/common/state/redis.cc. Changes to the hash function should only be
   * made through std::hash in src/common/common.h */
  std::hash<ray::UniqueID> hash;
  return PyLong_FromSize_t(hash(self->object_id));
}

static long PyObjectID_hash(PyObjectID *self) {
  // TODO(pcm): Replace this with a faster hash function. This currently
  // creates a tuple of length 20 and hashes it, which is slow
  PyObject *tuple = PyTuple_New(kUniqueIDSize);
  for (int i = 0; i < kUniqueIDSize; ++i) {
    PyTuple_SetItem(tuple, i, PyLong_FromLong(self->object_id.data()[i]));
  }
  long hash = PyObject_Hash(tuple);
  Py_XDECREF(tuple);
  return hash;
}

static PyObject *PyObjectID_repr(PyObjectID *self) {
  std::string repr = "ObjectID(" + self->object_id.hex() + ")";
  PyObject *result = PyUnicode_FromString(repr.c_str());
  return result;
}

static PyObject *PyObjectID___reduce__(PyObjectID *self) {
  PyErr_SetString(CommonError, "ObjectID objects cannot be serialized.");
  return NULL;
}

static PyMethodDef PyObjectID_methods[] = {
    {"id", (PyCFunction) PyObjectID_id, METH_NOARGS,
     "Return the hash associated with this ObjectID"},
    {"redis_shard_hash", (PyCFunction) PyObjectID_redis_shard_hash, METH_NOARGS,
     "Return the redis shard that this ObjectID is associated with"},
    {"hex", (PyCFunction) PyObjectID_hex, METH_NOARGS,
     "Return the object ID as a string in hex."},
    {"__reduce__", (PyCFunction) PyObjectID___reduce__, METH_NOARGS,
     "Say how to pickle this ObjectID. This raises an exception to prevent"
     "object IDs from being serialized."},
    {NULL} /* Sentinel */
};

static PyMemberDef PyObjectID_members[] = {
    {NULL} /* Sentinel */
};

PyTypeObject PyObjectIDType = {
    PyVarObject_HEAD_INIT(NULL, 0)        /* ob_size */
    "common.ObjectID",                    /* tp_name */
    sizeof(PyObjectID),                   /* tp_basicsize */
    0,                                    /* tp_itemsize */
    0,                                    /* tp_dealloc */
    0,                                    /* tp_print */
    0,                                    /* tp_getattr */
    0,                                    /* tp_setattr */
    0,                                    /* tp_compare */
    (reprfunc) PyObjectID_repr,           /* tp_repr */
    0,                                    /* tp_as_number */
    0,                                    /* tp_as_sequence */
    0,                                    /* tp_as_mapping */
    (hashfunc) PyObjectID_hash,           /* tp_hash */
    0,                                    /* tp_call */
    0,                                    /* tp_str */
    0,                                    /* tp_getattro */
    0,                                    /* tp_setattro */
    0,                                    /* tp_as_buffer */
    Py_TPFLAGS_DEFAULT,                   /* tp_flags */
    "ObjectID object",                    /* tp_doc */
    0,                                    /* tp_traverse */
    0,                                    /* tp_clear */
    (richcmpfunc) PyObjectID_richcompare, /* tp_richcompare */
    0,                                    /* tp_weaklistoffset */
    0,                                    /* tp_iter */
    0,                                    /* tp_iternext */
    PyObjectID_methods,                   /* tp_methods */
    PyObjectID_members,                   /* tp_members */
    0,                                    /* tp_getset */
    0,                                    /* tp_base */
    0,                                    /* tp_dict */
    0,                                    /* tp_descr_get */
    0,                                    /* tp_descr_set */
    0,                                    /* tp_dictoffset */
    (initproc) PyObjectID_init,           /* tp_init */
    0,                                    /* tp_alloc */
    PyType_GenericNew,                    /* tp_new */
};

/* Define the PyTask class. */

static int PyTask_init(PyTask *self, PyObject *args, PyObject *kwds) {
  /* ID of the driver that this task originates from. */
  UniqueID driver_id;
  /* ID of the actor this task should run on. */
  UniqueID actor_id = ActorID::nil();
  /* ID of the actor handle used to submit this task. */
  UniqueID actor_handle_id = ActorHandleID::nil();
  /* How many tasks have been launched on the actor so far? */
  int actor_counter = 0;
  /* True if this is an actor checkpoint task and false otherwise. */
  PyObject *is_actor_checkpoint_method_object = nullptr;
  /* ID of the function this task executes. */
  FunctionID function_id;
  /* Arguments of the task (can be PyObjectIDs or Python values). */
  PyObject *arguments;
  /* Number of return values of this task. */
  int num_returns;
  /* The ID of the task that called this task. */
  TaskID parent_task_id;
  /* The number of tasks that the parent task has called prior to this one. */
  int parent_counter;
  // The actor creation ID.
  ActorID actor_creation_id = ActorID::nil();
  // The dummy object for the actor creation task (if this is an actor method).
  ObjectID actor_creation_dummy_object_id = ObjectID::nil();
  /* Arguments of the task that are execution-dependent. These must be
   * PyObjectIDs). */
  PyObject *execution_arguments = nullptr;
  /* Dictionary of resource requirements for this task. */
  PyObject *resource_map = nullptr;
  // True if we should use the raylet code path and false otherwise.
  PyObject *use_raylet_object = nullptr;
  if (!PyArg_ParseTuple(
          args, "O&O&OiO&i|O&O&O&O&iOOOO", &PyObjectToUniqueID, &driver_id,
          &PyObjectToUniqueID, &function_id, &arguments, &num_returns,
          &PyObjectToUniqueID, &parent_task_id, &parent_counter,
          &PyObjectToUniqueID, &actor_creation_id, &PyObjectToUniqueID,
          &actor_creation_dummy_object_id, &PyObjectToUniqueID, &actor_id,
          &PyObjectToUniqueID, &actor_handle_id, &actor_counter,
          &is_actor_checkpoint_method_object, &execution_arguments,
          &resource_map, &use_raylet_object)) {
    return -1;
  }

  bool is_actor_checkpoint_method = false;
  if (is_actor_checkpoint_method_object != nullptr &&
      PyObject_IsTrue(is_actor_checkpoint_method_object) == 1) {
    is_actor_checkpoint_method = true;
  }

  // Parse the resource map.
  std::unordered_map<std::string, double> required_resources;

  bool found_CPU_requirements = false;
  PyObject *key, *value;
  Py_ssize_t position = 0;
  if (resource_map != nullptr) {
    if (!PyDict_Check(resource_map)) {
      PyErr_SetString(PyExc_TypeError, "resource_map must be a dictionary");
      return -1;
    }
    while (PyDict_Next(resource_map, &position, &key, &value)) {
      if (!(PyBytes_Check(key) || PyUnicode_Check(key))) {
        PyErr_SetString(PyExc_TypeError,
                        "the keys in resource_map must be strings");
        return -1;
      }
      if (!(PyFloat_Check(value) || PyInt_Check(value) ||
            PyLong_Check(value))) {
        PyErr_SetString(PyExc_TypeError,
                        "the values in resource_map must be floats");
        return -1;
      }
      // Handle the case where the key is a bytes object and the case where it
      // is a unicode object.
      std::string resource_name;
      if (PyUnicode_Check(key)) {
        PyObject *ascii_key = PyUnicode_AsASCIIString(key);
        resource_name =
            std::string(PyBytes_AsString(ascii_key), PyBytes_Size(ascii_key));
        Py_DECREF(ascii_key);
      } else {
        resource_name = std::string(PyBytes_AsString(key), PyBytes_Size(key));
      }
      if (resource_name == std::string("CPU")) {
        found_CPU_requirements = true;
      }
      required_resources[resource_name] = PyFloat_AsDouble(value);
    }
  }
  if (!found_CPU_requirements) {
    required_resources["CPU"] = 1.0;
  }

  Py_ssize_t num_args = PyList_Size(arguments);

  bool use_raylet = false;
  if (use_raylet_object != nullptr && PyObject_IsTrue(use_raylet_object) == 1) {
    use_raylet = true;
  }
  self->spec = nullptr;
  self->task_spec = nullptr;

  // Create the task spec.
  if (!use_raylet) {
    // The non-raylet code path.

    // Construct the task specification.
    TaskSpec_start_construct(
        g_task_builder, driver_id, parent_task_id, parent_counter,
        actor_creation_id, actor_creation_dummy_object_id, actor_id,
        actor_handle_id, actor_counter, is_actor_checkpoint_method, function_id,
        num_returns);
    // Add the task arguments.
    for (Py_ssize_t i = 0; i < num_args; ++i) {
      PyObject *arg = PyList_GetItem(arguments, i);
      if (PyObject_IsInstance(arg,
                              reinterpret_cast<PyObject *>(&PyObjectIDType))) {
        TaskSpec_args_add_ref(g_task_builder,
                              &(reinterpret_cast<PyObjectID *>(arg))->object_id,
                              1);
      } else {
        PyObject *data = PyObject_CallMethodObjArgs(pickle_module, pickle_dumps,
                                                    arg, pickle_protocol, NULL);
        TaskSpec_args_add_val(
            g_task_builder, reinterpret_cast<uint8_t *>(PyBytes_AsString(data)),
            PyBytes_Size(data));
        Py_DECREF(data);
      }
    }
    // Set the resource requirements for the task.
    for (auto const &resource_pair : required_resources) {
      TaskSpec_set_required_resource(g_task_builder, resource_pair.first,
                                     resource_pair.second);
    }

    // Compute the task ID and the return object IDs.
    self->spec = TaskSpec_finish_construct(g_task_builder, &self->size);

  } else {
    // The raylet code path.

    // Parse the arguments from the list.
    std::vector<std::shared_ptr<ray::raylet::TaskArgument>> args;
    for (Py_ssize_t i = 0; i < num_args; ++i) {
      PyObject *arg = PyList_GetItem(arguments, i);
      if (PyObject_IsInstance(arg,
                              reinterpret_cast<PyObject *>(&PyObjectIDType))) {
        std::vector<ObjectID> references = {
            reinterpret_cast<PyObjectID *>(arg)->object_id};
        args.push_back(
            std::make_shared<ray::raylet::TaskArgumentByReference>(references));
      } else {
        PyObject *data = PyObject_CallMethodObjArgs(pickle_module, pickle_dumps,
                                                    arg, pickle_protocol, NULL);
        args.push_back(std::make_shared<ray::raylet::TaskArgumentByValue>(
            reinterpret_cast<uint8_t *>(PyBytes_AsString(data)),
            PyBytes_Size(data)));
        Py_DECREF(data);
      }
    }

    self->task_spec = new ray::raylet::TaskSpecification(
        driver_id, parent_task_id, parent_counter, actor_creation_id,
        actor_creation_dummy_object_id, actor_id, actor_handle_id,
        actor_counter, function_id, args, num_returns, required_resources);
  }

  /* Set the task's execution dependencies. */
  self->execution_dependencies = new std::vector<ObjectID>();
  if (execution_arguments != NULL) {
    Py_ssize_t num_execution_args = PyList_Size(execution_arguments);
    for (Py_ssize_t i = 0; i < num_execution_args; ++i) {
      PyObject *execution_arg = PyList_GetItem(execution_arguments, i);
      if (!PyObject_IsInstance(execution_arg, (PyObject *) &PyObjectIDType)) {
        PyErr_SetString(PyExc_TypeError,
                        "Execution arguments must be an ObjectID.");
        return -1;
      }
      self->execution_dependencies->push_back(
          ((PyObjectID *) execution_arg)->object_id);
    }
  }

  return 0;
}

static void PyTask_dealloc(PyTask *self) {
  if (!use_raylet(self)) {
    TaskSpec_free(self->spec);
  } else {
    delete self->task_spec;
  }
  delete self->execution_dependencies;
  Py_TYPE(self)->tp_free(reinterpret_cast<PyObject *>(self));
}

static PyObject *PyTask_function_id(PyTask *self) {
  FunctionID function_id;
  if (!use_raylet(self)) {
    function_id = TaskSpec_function(self->spec);
  } else {
    function_id = self->task_spec->FunctionId();
  }
  return PyObjectID_make(function_id);
}

static PyObject *PyTask_actor_id(PyTask *self) {
  ActorID actor_id;
  if (!use_raylet(self)) {
    actor_id = TaskSpec_actor_id(self->spec);
  } else {
    actor_id = self->task_spec->ActorId();
  }
  return PyObjectID_make(actor_id);
}

static PyObject *PyTask_actor_counter(PyTask *self) {
  int64_t actor_counter;
  if (!use_raylet(self)) {
    actor_counter = TaskSpec_actor_counter(self->spec);
  } else {
    actor_counter = self->task_spec->ActorCounter();
  }
  return PyLong_FromLongLong(actor_counter);
}

static PyObject *PyTask_driver_id(PyTask *self) {
  UniqueID driver_id;
  if (!use_raylet(self)) {
    driver_id = TaskSpec_driver_id(self->spec);
  } else {
    driver_id = self->task_spec->DriverId();
  }
  return PyObjectID_make(driver_id);
}

static PyObject *PyTask_task_id(PyTask *self) {
  TaskID task_id;
  if (!use_raylet(self)) {
    task_id = TaskSpec_task_id(self->spec);
  } else {
    task_id = self->task_spec->TaskId();
  }
  return PyObjectID_make(task_id);
}

static PyObject *PyTask_parent_task_id(PyTask *self) {
  TaskID task_id;
  if (!use_raylet(self)) {
    task_id = TaskSpec_parent_task_id(self->spec);
  } else {
    task_id = self->task_spec->ParentTaskId();
  }
  return PyObjectID_make(task_id);
}

static PyObject *PyTask_parent_counter(PyTask *self) {
  int64_t parent_counter;
  if (!use_raylet(self)) {
    parent_counter = TaskSpec_parent_counter(self->spec);
  } else {
    parent_counter = self->task_spec->ParentCounter();
  }
  return PyLong_FromLongLong(parent_counter);
}

static PyObject *PyTask_arguments(PyTask *self) {
  TaskSpec *task = self->spec;
  ray::raylet::TaskSpecification *task_spec = self->task_spec;

  int64_t num_args;
  if (!use_raylet(self)) {
    num_args = TaskSpec_num_args(task);
  } else {
    num_args = self->task_spec->NumArgs();
  }

  PyObject *arg_list = PyList_New((Py_ssize_t) num_args);
  for (int i = 0; i < num_args; ++i) {
    int count;
    if (!use_raylet(self)) {
      count = TaskSpec_arg_id_count(task, i);
    } else {
      count = task_spec->ArgIdCount(i);
    }

    if (count > 0) {
      assert(count == 1);

      ObjectID object_id;
      if (!use_raylet(self)) {
        object_id = TaskSpec_arg_id(task, i, 0);
      } else {
        object_id = task_spec->ArgId(i, 0);
      }

      PyList_SetItem(arg_list, i, PyObjectID_make(object_id));
    } else {
      RAY_CHECK(pickle_module != NULL);
      RAY_CHECK(pickle_loads != NULL);

      const uint8_t *arg_val;
      int64_t arg_length;
      if (!use_raylet(self)) {
        arg_val = TaskSpec_arg_val(task, i);
        arg_length = TaskSpec_arg_length(task, i);
      } else {
        arg_val = task_spec->ArgVal(i);
        arg_length = task_spec->ArgValLength(i);
      }

      PyObject *str =
          PyBytes_FromStringAndSize(reinterpret_cast<const char *>(arg_val),
                                    static_cast<Py_ssize_t>(arg_length));
      PyObject *val =
          PyObject_CallMethodObjArgs(pickle_module, pickle_loads, str, NULL);
      Py_XDECREF(str);
      PyList_SetItem(arg_list, i, val);
    }
  }
  return arg_list;
}

static PyObject *PyTask_actor_creation_id(PyTask *self) {
  ActorID actor_creation_id;
  if (!use_raylet(self)) {
    actor_creation_id = TaskSpec_actor_creation_id(self->spec);
  } else {
    actor_creation_id = self->task_spec->ActorCreationId();
  }
  return PyObjectID_make(actor_creation_id);
}

static PyObject *PyTask_actor_creation_dummy_object_id(PyTask *self) {
  ObjectID actor_creation_dummy_object_id;
  if (!use_raylet(self)) {
    if (TaskSpec_is_actor_task(self->spec)) {
      actor_creation_dummy_object_id =
          TaskSpec_actor_creation_dummy_object_id(self->spec);
    } else {
      actor_creation_dummy_object_id = ObjectID::nil();
    }
  } else {
    actor_creation_dummy_object_id =
        self->task_spec->ActorCreationDummyObjectId();
  }
  return PyObjectID_make(actor_creation_dummy_object_id);
}

static PyObject *PyTask_required_resources(PyTask *self) {
  PyObject *required_resources = PyDict_New();

  std::unordered_map<std::string, double> resource_map;
  if (!use_raylet(self)) {
    resource_map = TaskSpec_get_required_resources(self->spec);
  } else {
    resource_map = self->task_spec->GetRequiredResources().GetResourceMap();
  }

  for (auto const &resource_pair : resource_map) {
    std::string resource_name = resource_pair.first;
#if PY_MAJOR_VERSION >= 3
    PyObject *key =
        PyUnicode_FromStringAndSize(resource_name.data(), resource_name.size());
#else
    PyObject *key =
        PyBytes_FromStringAndSize(resource_name.data(), resource_name.size());
#endif
    PyObject *value = PyFloat_FromDouble(resource_pair.second);
    PyDict_SetItem(required_resources, key, value);
    Py_DECREF(key);
    Py_DECREF(value);
  }
  return required_resources;
}

static PyObject *PyTask_returns(PyTask *self) {
  TaskSpec *task = self->spec;
  ray::raylet::TaskSpecification *task_spec = self->task_spec;

  int64_t num_returns;
  if (!use_raylet(self)) {
    num_returns = TaskSpec_num_returns(task);
  } else {
    num_returns = task_spec->NumReturns();
  }

  PyObject *return_id_list = PyList_New((Py_ssize_t) num_returns);
  for (int i = 0; i < num_returns; ++i) {
    ObjectID object_id;
    if (!use_raylet(self)) {
      object_id = TaskSpec_return(task, i);
    } else {
      object_id = task_spec->ReturnId(i);
    }
    PyList_SetItem(return_id_list, i, PyObjectID_make(object_id));
  }
  return return_id_list;
}

static PyObject *PyTask_execution_dependencies_string(PyTask *self) {
  flatbuffers::FlatBufferBuilder fbb;
  auto execution_dependencies = CreateTaskExecutionDependencies(
      fbb, to_flatbuf(fbb, *self->execution_dependencies));
  fbb.Finish(execution_dependencies);
  return PyBytes_FromStringAndSize((char *) fbb.GetBufferPointer(),
                                   fbb.GetSize());
}

static PyMethodDef PyTask_methods[] = {
    {"function_id", (PyCFunction) PyTask_function_id, METH_NOARGS,
     "Return the function ID for this task."},
    {"parent_task_id", (PyCFunction) PyTask_parent_task_id, METH_NOARGS,
     "Return the task ID of the parent task."},
    {"parent_counter", (PyCFunction) PyTask_parent_counter, METH_NOARGS,
     "Return the parent counter of this task."},
    {"actor_id", (PyCFunction) PyTask_actor_id, METH_NOARGS,
     "Return the actor ID for this task."},
    {"actor_counter", (PyCFunction) PyTask_actor_counter, METH_NOARGS,
     "Return the actor counter for this task."},
    {"driver_id", (PyCFunction) PyTask_driver_id, METH_NOARGS,
     "Return the driver ID for this task."},
    {"task_id", (PyCFunction) PyTask_task_id, METH_NOARGS,
     "Return the task ID for this task."},
    {"arguments", (PyCFunction) PyTask_arguments, METH_NOARGS,
     "Return the arguments for the task."},
    {"actor_creation_id", (PyCFunction) PyTask_actor_creation_id, METH_NOARGS,
     "Return the actor creation ID for the task."},
    {"actor_creation_dummy_object_id",
     (PyCFunction) PyTask_actor_creation_dummy_object_id, METH_NOARGS,
     "Return the actor creation dummy object ID for the task."},
    {"required_resources", (PyCFunction) PyTask_required_resources, METH_NOARGS,
     "Return the resource vector of the task."},
    {"returns", (PyCFunction) PyTask_returns, METH_NOARGS,
     "Return the object IDs for the return values of the task."},
    {"execution_dependencies_string",
     (PyCFunction) PyTask_execution_dependencies_string, METH_NOARGS,
     "Return the execution dependencies for the task as a string."},
    {NULL} /* Sentinel */
};

PyTypeObject PyTaskType = {
    PyVarObject_HEAD_INIT(NULL, 0) /* ob_size */
    "task.Task",                   /* tp_name */
    sizeof(PyTask),                /* tp_basicsize */
    0,                             /* tp_itemsize */
    (destructor) PyTask_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 */
    "Task object",                 /* tp_doc */
    0,                             /* tp_traverse */
    0,                             /* tp_clear */
    0,                             /* tp_richcompare */
    0,                             /* tp_weaklistoffset */
    0,                             /* tp_iter */
    0,                             /* tp_iternext */
    PyTask_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) PyTask_init,        /* tp_init */
    0,                             /* tp_alloc */
    PyType_GenericNew,             /* tp_new */
};

/* Create a PyTask from a C struct. The resulting PyTask takes ownership of the
 * TaskSpec and will deallocate the TaskSpec in the PyTask destructor. */
PyObject *PyTask_make(TaskSpec *task_spec, int64_t task_size) {
  PyTask *result = PyObject_New(PyTask, &PyTaskType);
  result = (PyTask *) PyObject_Init((PyObject *) result, &PyTaskType);
  result->spec = task_spec;
  result->size = task_size;
  /* The created task does not include any execution dependencies. */
  result->execution_dependencies = new std::vector<ObjectID>();
  return (PyObject *) result;
}

/* Define the methods for the module. */

/**
 * This method checks if a Python object is sufficiently simple that it can be
 * serialized and passed by value as an argument to a task (without being put in
 * the object store). The details of which objects are sufficiently simple are
 * defined by this method and are not particularly important. But for
 * performance reasons, it is better to place "small" objects in the task itself
 * and "large" objects in the object store.
 *
 * @param value The Python object in question.
 * @param num_elements_contained If this method returns 1, then the number of
 *        objects recursively contained within this object will be added to the
 *        value at this address. This is used to make sure that we do not
 *        serialize objects that are too large.
 * @return False if the object cannot be serialized in the task and true if it
 *         can.
 */
bool is_simple_value(PyObject *value, int *num_elements_contained) {
  *num_elements_contained += 1;
  if (*num_elements_contained >= RayConfig::instance().num_elements_limit()) {
    return false;
  }
  if (PyInt_Check(value) || PyLong_Check(value) || value == Py_False ||
      value == Py_True || PyFloat_Check(value) || value == Py_None) {
    return true;
  }
  if (PyBytes_CheckExact(value)) {
    *num_elements_contained += PyBytes_Size(value);
    return (*num_elements_contained <
            RayConfig::instance().num_elements_limit());
  }
  if (PyUnicode_CheckExact(value)) {
    *num_elements_contained += PyUnicode_GET_SIZE(value);
    return (*num_elements_contained <
            RayConfig::instance().num_elements_limit());
  }
  if (PyList_CheckExact(value) &&
      PyList_Size(value) < RayConfig::instance().size_limit()) {
    for (Py_ssize_t i = 0; i < PyList_Size(value); ++i) {
      if (!is_simple_value(PyList_GetItem(value, i), num_elements_contained)) {
        return false;
      }
    }
    return (*num_elements_contained <
            RayConfig::instance().num_elements_limit());
  }
  if (PyDict_CheckExact(value) &&
      PyDict_Size(value) < RayConfig::instance().size_limit()) {
    PyObject *key, *val;
    Py_ssize_t pos = 0;
    while (PyDict_Next(value, &pos, &key, &val)) {
      if (!is_simple_value(key, num_elements_contained) ||
          !is_simple_value(val, num_elements_contained)) {
        return false;
      }
    }
    return (*num_elements_contained <
            RayConfig::instance().num_elements_limit());
  }
  if (PyTuple_CheckExact(value) &&
      PyTuple_Size(value) < RayConfig::instance().size_limit()) {
    for (Py_ssize_t i = 0; i < PyTuple_Size(value); ++i) {
      if (!is_simple_value(PyTuple_GetItem(value, i), num_elements_contained)) {
        return false;
      }
    }
    return (*num_elements_contained <
            RayConfig::instance().num_elements_limit());
  }
  if (PyArray_CheckExact(value)) {
    PyArrayObject *array = reinterpret_cast<PyArrayObject *>(value);
    if (PyArray_TYPE(array) == NPY_OBJECT) {
      return false;
    }
    *num_elements_contained += PyArray_NBYTES(array);
    return (*num_elements_contained <
            RayConfig::instance().num_elements_limit());
  }
  return false;
}

PyObject *check_simple_value(PyObject *self, PyObject *args) {
  PyObject *value;
  if (!PyArg_ParseTuple(args, "O", &value)) {
    return NULL;
  }
  int num_elements_contained = 0;
  if (is_simple_value(value, &num_elements_contained)) {
    Py_RETURN_TRUE;
  }
  Py_RETURN_FALSE;
}