+
+
improv allows users to flexibly specify and manage adaptive experiments to integrate data collection, preprocessing, visualization, and user-defined analytics. All kinds of behavioral, neural, or modeling data can be incorporated, and input and output data streams are managed independently and asynchronously. With this design, streaming analyses and real-time interventions can be easily integrated into various experimental setups. improv manages the backend engineering of data flow and task execution for all steps in an experimental pipeline in real time, without requiring user oversight. Users need only define their particular processing pipeline with simple text files and are free to define their own streaming analyses via Python classes, allowing for rapid prototyping of adaptive experiments.
+
-improv's design is based on a steamlined version of the actor model for concurrent computation. Each component of the system (experimental pipeline) is considered an 'actor' and has a unique role. They interact via message passing, without the need for a central broker. Actors are implemented as user-defined classes that inherit from improv's Actor class, which supplies all queues for message passing and orchestrates process execution and error handling. Messages between actors are composed of keys that correspond to items in a shared, in-memory data store. This both minimizes communication overhead and data copying between processes.
+_improv_'s design is based on a streamlined version of the actor model for concurrent computation. Each component of the system (experimental pipeline) is considered an 'actor' and has a unique role. They interact via message passing, without the need for a central broker. Actors are implemented as user-defined classes that inherit from _improv_'s `Actor` class, which supplies all queues for message passing and orchestrates process execution and error handling. Messages between actors are composed of keys that correspond to items in a shared, in-memory data store. This both minimizes communication overhead and data copying between processes.
## Installation
-For installation instructions, please consult the [wiki](https://github.com/project-improv/improv/wiki/Installation) on our github.
+For installation instructions, please consult the [docs](https://project-improv.github.io/improv/installation.html) on our github.
### Contact
-To get in touch, feel free to reach out on Twitter @annedraelos.
+To get in touch, feel free to reach out on Twitter @annedraelos or @jmxpearson.
diff --git a/demos/bubblewrap/README.md b/demos/bubblewrap/README.md
index 11c9830d2..5b570d245 100644
--- a/demos/bubblewrap/README.md
+++ b/demos/bubblewrap/README.md
@@ -1,6 +1,6 @@
# Running Bubblewrap demo
## Ubuntu users
-After installing improv, install additional dependencies (JAX on CPU by default) navigating to `demos/bubblewrap` and doing
+After installing _improv_, install additional dependencies (JAX on CPU by default) navigating to `demos/bubblewrap` and doing
- `pip install -r requirements.txt`
diff --git a/demos/minimal/README.md b/demos/minimal/README.md
new file mode 100644
index 000000000..4e3445ad0
--- /dev/null
+++ b/demos/minimal/README.md
@@ -0,0 +1,67 @@
+# minimal demo
+
+This folder contains a minimal demo for running improv. In this demo, data generated by the **generator** `actor` is
+stored in a data store and a key to the generated data is sent via a queue to the **processor** `actor` that accesses
+the data in the store and processes it.
+
+Usage:
+
+```bash
+# cd to this dir
+cd .../improv/demos/minimal
+
+# start improv
+improv run ./minimal.yaml
+
+# call `setup` in the improv TUI
+setup
+
+# call `run` in the improv TUI
+run
+
+# when you are ready to stop the process, call `stop` in the improv TUI
+stop
+```
+
+## Visualization using `fastplotlib`
+
+You can also run the minimal demo and visualize the generated data using `fastplotlib`.
+
+As before, data is generated by the **generator** `actor` is stored in a data store and a key to the generated data is
+sent via a queue to the **processor** `actor` that accesses the data in the store and processes it. Additionally, the
+`fastplotlib.ipynb` notebook then receives the most recent data via `zmq` and displays it using
+[`fastplotlib`](https://github.com/fastplotlib/fastplotlib).
+
+### Instructions
+
+1. Swap the **processor** `actor` found in `/improv/demos/sample_actors/visual/sample_processor.py` for the one in
+the minimal demo
+
+2. Run `pip install -r requirements.txt` in this directory.
+
+Usage:
+
+```bash
+# cd to this dir
+cd .../improv/demos/minimal
+
+# start improv
+improv run ./minimal.yaml
+
+# call `setup` in the improv TUI
+setup
+
+# Run the cells in the jupyter notebook until you receive
+# a plot that has a white cosine wave
+
+# once the plot is ready call `run` in the improv TUI
+run
+
+# You should see the plot updating between a cosine and sine wave with different colors depending on
+# whether the frame number is even or odd
+
+# when you are ready to stop the process, call `stop` in the improv TUI
+stop
+```
+
+#### Note: The `fastplotlib.ipynb` can only be run in `jupyter lab`
\ No newline at end of file
diff --git a/demos/minimal/actors/sample_generator.py b/demos/minimal/actors/sample_generator.py
index df5191ec5..313358330 100644
--- a/demos/minimal/actors/sample_generator.py
+++ b/demos/minimal/actors/sample_generator.py
@@ -1,5 +1,4 @@
-from improv.actor import Actor, RunManager
-from datetime import date # used for saving
+from improv.actor import Actor
import numpy as np
import logging
@@ -50,13 +49,20 @@ def runStep(self):
"""
if self.frame_num < np.shape(self.data)[0]:
- data_id = self.client.put(
- self.data[self.frame_num], str(f"Gen_raw: {self.frame_num}")
- )
+ if self.store_loc:
+ data_id = self.client.put(
+ self.data[self.frame_num], str(f"Gen_raw: {self.frame_num}")
+ )
+ else:
+ data_id = self.client.put(self.data[self.frame_num])
# logger.info('Put data in store')
try:
- self.q_out.put([[data_id, str(self.frame_num)]])
- logger.info("Sent message on")
+ if self.store_loc:
+ self.q_out.put([[data_id, str(self.frame_num)]])
+ else:
+ self.q_out.put(data_id)
+ # logger.info("Sent message on")
+
self.frame_num += 1
except Exception as e:
logger.error(
diff --git a/demos/minimal/actors/sample_processor.py b/demos/minimal/actors/sample_processor.py
index 24f87880d..feeb9f688 100644
--- a/demos/minimal/actors/sample_processor.py
+++ b/demos/minimal/actors/sample_processor.py
@@ -45,19 +45,23 @@ def runStep(self):
frame = None
try:
- frame = self.q_in.get(timeout=0.001)
+ frame = self.q_in.get(timeout=0.05)
- except:
+ except Exception:
logger.error("Could not get frame!")
pass
if frame is not None and self.frame_num is not None:
self.done = False
- self.frame = self.client.getID(frame[0][0])
+ if self.store_loc:
+ self.frame = self.client.getID(frame[0][0])
+ else:
+ self.frame = self.client.get(frame)
avg = np.mean(self.frame[0])
- # print(f"Average: {avg}")
+ logger.info(f"Average: {avg}")
self.avg_list.append(avg)
- # print(f"Overall Average: {np.mean(self.avg_list)}")
- # print(f"Frame number: {self.frame_num}")
- self.frame_num += 1
+ logger.info(f"Overall Average: {np.mean(self.avg_list)}")
+ logger.info(f"Frame number: {self.frame_num}")
+
+ self.frame_num += 1
\ No newline at end of file
diff --git a/demos/minimal/actors/sample_spawn_processor.py b/demos/minimal/actors/sample_spawn_processor.py
index 96b031956..cdc8bbb73 100644
--- a/demos/minimal/actors/sample_spawn_processor.py
+++ b/demos/minimal/actors/sample_spawn_processor.py
@@ -1,4 +1,4 @@
-from improv.actor import Actor, RunManager
+from improv.actor import Actor
import numpy as np
from queue import Empty
import logging
@@ -39,17 +39,6 @@ def stop(self):
logger.info("Processor stopping")
return 0
- # def run(self):
- # """ Send array into the store.
- # """
- # self.fcns = {}
- # self.fcns['setup'] = self.setup
- # self.fcns['run'] = self.runStep
- # self.fcns['stop'] = self.stop
-
- # with RunManager(self.name, self.fcns, self.links) as rm:
- # logger.info(rm)
-
def runStep(self):
"""Gets from the input queue and calculates the average.
@@ -63,16 +52,21 @@ def runStep(self):
frame = self.q_in.get(timeout=0.05)
except Empty:
pass
- except:
+ except Exception:
logger.error("Could not get frame!")
pass
if frame is not None and self.frame_num is not None:
self.done = False
- self.frame = self.client.getID(frame[0][0])
+ if self.store_loc:
+ self.frame = self.client.getID(frame[0][0])
+ else:
+ self.frame = self.client.get(frame)
avg = np.mean(self.frame[0])
- print(f"Average: {avg}")
+
+ logger.info(f"Average: {avg}")
self.avg_list.append(avg)
- print(f"Overall Average: {np.mean(self.avg_list)}")
- print(f"Frame number: {self.frame_num}")
+ logger.info(f"Overall Average: {np.mean(self.avg_list)}")
+ logger.info(f"Frame number: {self.frame_num}")
+
self.frame_num += 1
diff --git a/demos/minimal/minimal.yaml b/demos/minimal/minimal.yaml
index 230282d17..c8e0b24ee 100644
--- a/demos/minimal/minimal.yaml
+++ b/demos/minimal/minimal.yaml
@@ -8,4 +8,7 @@ actors:
class: Processor
connections:
- Generator.q_out: [Processor.q_in]
\ No newline at end of file
+ Generator.q_out: [Processor.q_in]
+
+redis_config:
+ port: 6379
\ No newline at end of file
diff --git a/demos/minimal/minimal_plasma.yaml b/demos/minimal/minimal_plasma.yaml
new file mode 100644
index 000000000..ab8696783
--- /dev/null
+++ b/demos/minimal/minimal_plasma.yaml
@@ -0,0 +1,13 @@
+actors:
+ Generator:
+ package: actors.sample_generator
+ class: Generator
+
+ Processor:
+ package: actors.sample_processor
+ class: Processor
+
+connections:
+ Generator.q_out: [Processor.q_in]
+
+plasma_config:
\ No newline at end of file
diff --git a/demos/minimal/minimal_spawn.yaml b/demos/minimal/minimal_spawn.yaml
index 184bc18b2..6f22e653f 100644
--- a/demos/minimal/minimal_spawn.yaml
+++ b/demos/minimal/minimal_spawn.yaml
@@ -10,3 +10,6 @@ actors:
connections:
Generator.q_out: [Processor.q_in]
+
+redis_config:
+ port: 6378
\ No newline at end of file
diff --git a/demos/minimal/requirements.txt b/demos/minimal/requirements.txt
new file mode 100644
index 000000000..053362891
--- /dev/null
+++ b/demos/minimal/requirements.txt
@@ -0,0 +1,2 @@
+fastplotlib[notebook]
+simplejpeg
\ No newline at end of file
diff --git a/demos/sample_actors/analysis.py b/demos/sample_actors/analysis.py
index c754227e0..664ddaf31 100644
--- a/demos/sample_actors/analysis.py
+++ b/demos/sample_actors/analysis.py
@@ -16,8 +16,8 @@
class MeanAnalysis(Actor):
# TODO: Add additional error handling
# TODO: this is too complex for a sample actor?
- def __init__(self, *args):
- super().__init__(*args)
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
def setup(self, param_file=None):
"""Set custom parameters here
diff --git a/demos/sample_actors/analysis_async.py b/demos/sample_actors/analysis_async.py
index a6fcfa663..93a7888c5 100755
--- a/demos/sample_actors/analysis_async.py
+++ b/demos/sample_actors/analysis_async.py
@@ -21,8 +21,8 @@ class AnalysisAsync(Analysis):
"""
- def __init__(self, *args):
- super().__init__(*args)
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
self.frame_number = 0
self.aqueue = None
diff --git a/demos/sample_actors/analysis_julia.py b/demos/sample_actors/analysis_julia.py
index 56d8c9411..83d5b1d09 100644
--- a/demos/sample_actors/analysis_julia.py
+++ b/demos/sample_actors/analysis_julia.py
@@ -16,10 +16,10 @@ class JuliaAnalysis(Actor):
This actor puts in the data store the average frame intensity.
"""
- def __init__(self, *args, julia_file="julia_func.jl"):
+ def __init__(self, *args, julia_file="julia_func.jl", **kwargs):
"""julia_file: path to .jl file(s) for analyses computed in Julia"""
- super().__init__(*args)
+ super().__init__(*args, **kwargs)
self.julia = None
self.julia_file = julia_file
diff --git a/demos/sample_actors/process.py b/demos/sample_actors/process.py
index 78c7aa6d9..603064c81 100644
--- a/demos/sample_actors/process.py
+++ b/demos/sample_actors/process.py
@@ -29,9 +29,13 @@ class CaimanProcessor(Actor):
"""
def __init__(
- self, *args, init_filename="data/Tolias_mesoscope_2.hdf5", config_file=None
+ self,
+ *args,
+ init_filename="data/Tolias_mesoscope_2.hdf5",
+ config_file=None,
+ **kwargs,
):
- super().__init__(*args)
+ super().__init__(*args, **kwargs)
logger.info("initfile {}, config file {}".format(init_filename, config_file))
self.param_file = config_file
self.init_filename = init_filename
@@ -73,7 +77,9 @@ def setup(self):
self.counter = 0
def stop(self):
- print("Processor broke, avg time per frame: ", np.mean(self.total_times, axis=0))
+ print(
+ "Processor broke, avg time per frame: ", np.mean(self.total_times, axis=0)
+ )
print("Processor got through ", self.frame_number, " frames")
np.savetxt("output/timing/process_frame_time.txt", np.array(self.total_times))
np.savetxt("output/timing/process_timestamp.txt", np.array(self.timestamp))
@@ -85,7 +91,9 @@ def stop(self):
np.savetxt("output/timing/shape_time.txt", self.shape_time)
np.savetxt("output/timing/detect_time.txt", self.detect_time)
- np.savetxt("output/timing/putAnalysis_time.txt", np.array(self.putAnalysis_time))
+ np.savetxt(
+ "output/timing/putAnalysis_time.txt", np.array(self.putAnalysis_time)
+ )
np.savetxt("output/timing/procFrame_time.txt", np.array(self.procFrame_time))
print("Number of times coords updated ", self.counter)
@@ -93,11 +101,16 @@ def stop(self):
if self.onAc.estimates.OASISinstances is not None:
try:
init = self.params["init_batch"]
- S = np.stack([osi.s[init:] for osi in self.onAc.estimates.OASISinstances])
+ S = np.stack(
+ [osi.s[init:] for osi in self.onAc.estimates.OASISinstances]
+ )
np.savetxt("output/end_spikes.txt", S)
except Exception as e:
- logger.error("Exception {}: {} during frame number {}"
- .format(type(e).__name__, e, self.frame_number))
+ logger.error(
+ "Exception {}: {} during frame number {}".format(
+ type(e).__name__, e, self.frame_number
+ )
+ )
print(traceback.format_exc())
else:
print("No OASIS")
@@ -125,13 +138,18 @@ def runStep(self):
try:
self.frame = self.client.getID(frame[0][str(self.frame_number)])
t2 = time.time()
- self._fitFrame(self.frame_number + init, self.frame.reshape(-1, order="F"))
+ self._fitFrame(
+ self.frame_number + init, self.frame.reshape(-1, order="F")
+ )
self.fitframe_time.append([time.time() - t2])
self.putEstimates()
self.timestamp.append([time.time(), self.frame_number])
except ObjectNotFoundError:
- logger.error("Processor: Frame {} unavailable from store, droppping"
- .format(self.frame_number))
+ logger.error(
+ "Processor: Frame {} unavailable from store, droppping".format(
+ self.frame_number
+ )
+ )
self.dropped_frames.append(self.frame_number)
self.q_out.put([1])
except KeyError as e:
@@ -139,8 +157,11 @@ def runStep(self):
# Proceed at all costs
self.dropped_frames.append(self.frame_number)
except Exception as e:
- logger.error("Processor error: {}: {} during frame number {}"
- .format(type(e).__name__, e, self.frame_number))
+ logger.error(
+ "Processor error: {}: {} during frame number {}".format(
+ type(e).__name__, e, self.frame_number
+ )
+ )
print(traceback.format_exc())
self.dropped_frames.append(self.frame_number)
self.frame_number += 1
@@ -207,9 +228,11 @@ def putEstimates(self):
t = time.time()
nb = self.onAc.params.get("init", "nb")
A = self.onAc.estimates.Ab[:, nb:]
- before = self.params["init_batch"]
+ before = self.params["init_batch"]
# self.frame_number-500 if self.frame_number > 500 else 0
- C = self.onAc.estimates.C_on[nb : self.onAc.M, before : self.frame_number + before] # .get_ordered()
+ C = self.onAc.estimates.C_on[
+ nb : self.onAc.M, before : self.frame_number + before
+ ] # .get_ordered()
t2 = time.time()
t3 = time.time()
@@ -232,7 +255,9 @@ def putEstimates(self):
# self.q_comm.put([self.frame_number])
- self.putAnalysis_time.append([time.time() - t, t2 - t, t3 - t2, t4 - t3, t5 - t4, t6 - t5])
+ self.putAnalysis_time.append(
+ [time.time() - t, t2 - t, t3 - t2, t4 - t3, t5 - t4, t6 - t5]
+ )
def _checkFrames(self):
"""Check to see if we have frames for processing"""
@@ -281,13 +306,21 @@ def makeImage(self):
try:
# components = self.onAc.estimates.Ab[:,mn:].dot(self.onAc.estimates.C_on[mn:self.onAc.M,(self.frame_number-1)%self.onAc.window]).reshape(self.onAc.dims, order='F')
# background = self.onAc.estimates.Ab[:,:mn].dot(self.onAc.estimates.C_on[:mn,(self.frame_number-1)%self.onAc.window]).reshape(self.onAc.dims, order='F')
- components = (self.onAc.estimates.Ab[:, mn:]
- .dot(self.onAc.estimates.C_on[mn : self.onAc.M, (self.frame_number - 1)])
- .reshape(self.onAc.dims, order="F"))
- background = (self.onAc.estimates.Ab[:, :mn]
+ components = (
+ self.onAc.estimates.Ab[:, mn:]
+ .dot(
+ self.onAc.estimates.C_on[mn : self.onAc.M, (self.frame_number - 1)]
+ )
+ .reshape(self.onAc.dims, order="F")
+ )
+ background = (
+ self.onAc.estimates.Ab[:, :mn]
.dot(self.onAc.estimates.C_on[:mn, (self.frame_number - 1)])
- .reshape(self.onAc.dims, order="F"))
- image = ((components + background) - self.onAc.bnd_Y[0]) / np.diff(self.onAc.bnd_Y)
+ .reshape(self.onAc.dims, order="F")
+ )
+ image = ((components + background) - self.onAc.bnd_Y[0]) / np.diff(
+ self.onAc.bnd_Y
+ )
image = np.minimum((image * 255.0), 255).astype("u1")
except ValueError as ve:
logger.info("ValueError: {0}".format(ve))
diff --git a/demos/sample_actors/simple_analysis.py b/demos/sample_actors/simple_analysis.py
index 6cdd06173..ff350b583 100644
--- a/demos/sample_actors/simple_analysis.py
+++ b/demos/sample_actors/simple_analysis.py
@@ -14,8 +14,8 @@
class SimpleAnalysis(Actor):
# TODO: Add additional error handling
- def __init__(self, *args):
- super().__init__(*args)
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
def setup(self, param_file=None):
"""Set custom parameters here
diff --git a/demos/sample_actors/visual/fastplotlib.ipynb b/demos/sample_actors/visual/fastplotlib.ipynb
new file mode 100644
index 000000000..5fc7fde2a
--- /dev/null
+++ b/demos/sample_actors/visual/fastplotlib.ipynb
@@ -0,0 +1,369 @@
+{
+ "cells": [
+ {
+ "cell_type": "markdown",
+ "id": "83566503-c31b-4d83-b6f0-173bc9b2102d",
+ "metadata": {},
+ "source": [
+ "## `fastplotlib` demo"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 1,
+ "id": "initial_id",
+ "metadata": {},
+ "outputs": [
+ {
+ "data": {
+ "application/vnd.jupyter.widget-view+json": {
+ "model_id": "bd836038b56f4721890cd69530019350",
+ "version_major": 2,
+ "version_minor": 0
+ },
+ "text/plain": [
+ "Image(value=b'\\x89PNG\\r\\n\\x1a\\n\\x00\\x00\\x00\\rIHDR\\x00\\x00\\x01,\\x00\\x00\\x007\\x08\\x06\\x00\\x00\\x00\\xb6\\x1bw\\x99\\x…"
+ ]
+ },
+ "metadata": {},
+ "output_type": "display_data"
+ },
+ {
+ "name": "stderr",
+ "output_type": "stream",
+ "text": [
+ "MESA-INTEL: warning: cannot initialize blitter engine\n",
+ "Detected skylake derivative running on mesa i915. Clears to srgb textures will use manual shader clears.\n",
+ "Detected skylake derivative running on mesa i915. Clears to srgb textures will use manual shader clears.\n",
+ "Detected skylake derivative running on mesa i915. Clears to srgb textures will use manual shader clears.\n",
+ "Detected skylake derivative running on mesa i915. Clears to srgb textures will use manual shader clears.\n",
+ "Detected skylake derivative running on mesa i915. Clears to srgb textures will use manual shader clears.\n"
+ ]
+ },
+ {
+ "name": "stdout",
+ "output_type": "stream",
+ "text": [
+ "Available devices:\n",
+ "🯄 (default) | Intel(R) Arc(tm) Graphics (MTL) | IntegratedGPU | Vulkan | Mesa 24.0.8-1\n",
+ "❗ | llvmpipe (LLVM 17.0.6, 256 bits) | CPU | Vulkan | Mesa 24.0.8-1 (LLVM 17.0.6)\n",
+ "❗ | Mesa Intel(R) Arc(tm) Graphics (MTL) | IntegratedGPU | OpenGL | \n"
+ ]
+ }
+ ],
+ "source": [
+ "import numpy as np\n",
+ "import zmq\n",
+ "import fastplotlib as fpl"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "74e6f9a5-e4f3-4033-9529-20fcf5068703",
+ "metadata": {},
+ "source": [
+ "### Check if GPU or rendering with CPU via lavapipe is available"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 2,
+ "id": "093c8fd4-bbf8-4374-a661-494a61a73cca",
+ "metadata": {},
+ "outputs": [
+ {
+ "name": "stderr",
+ "output_type": "stream",
+ "text": [
+ "Detected skylake derivative running on mesa i915. Clears to srgb textures will use manual shader clears.\n",
+ "Detected skylake derivative running on mesa i915. Clears to srgb textures will use manual shader clears.\n"
+ ]
+ }
+ ],
+ "source": [
+ "if len(fpl.utils.gpu.enumerate_adapters()) < 1:\n",
+ " raise IndexError(\"WGPU could not enumerate any adapters, fastplotlib will not work.\")"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "24f4a955-c7c5-4b9d-a653-d5daecb27d04",
+ "metadata": {},
+ "source": [
+ "### Setup zmq subscriber client"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 3,
+ "id": "b4182903-b4fe-462a-9f92-71d254bb5e24",
+ "metadata": {},
+ "outputs": [
+ {
+ "data": {
+ "text/plain": [
+ "![](\"https://dibs-web01.vm.duke.edu/pearson/assets/improv/improv_quick_start.gif\")
\n",
+ "\n", + "
\n", + " " + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Details for running each actor can be found in the `README` file within each demo folder. \n", + "\n", + "````{warning}\n", + "Many demos have additional dependencies that are not part of the typical _improv_ installation. In these cases, additional packages may be installed by running\n", + "```bash\n", + "pip install -r requirements.txt\n", + "```\n", + "within the demo folder. Again, see the individual README files within each demo folder for details.\n", + "````" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "improv", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.11.3" + } + }, + "nbformat": 4, + "nbformat_minor": 2 +} diff --git a/docs/design.md b/docs/design.md new file mode 100644 index 000000000..9bd6b188d --- /dev/null +++ b/docs/design.md @@ -0,0 +1,101 @@ +(page:design)= +# _improv_'s design + +Here, we cover some technical details that can be useful in understanding how _improv_ works under the hood. + +One way of viewing _improv_ is as a lightweight networking library that handles three key tasks: +1. **Pipeline specification and setup.** Experimental pipelines are defined by YAML files. At run time, _improv_ reads these files and starts separate system processes corresponding to each processing step (actor). +1. **Interprocess communication.** As part of each experiment, data need to be collected and passed between processes. _improv_ uses a centralized server and in-memory data store to coordinate this process. +1. **Logging and data persistence.** After the experiment is done, we need records of both what happened during the experiment and what data were collected. _improv_ organizes both of these processes by creating a unified log file and storing data to disk as the experiment runs. + +We consider each of these components below. + +(page:design:pipeline_spec)= +## Pipeline specification + +_improv_ pipelines are [directed graphs](https://en.wikipedia.org/wiki/Directed_graph), with each node corresponding to a processing step (called an "actor" after the [actor model](https://en.wikipedia.org/wiki/Actor_model) of concurrency {cite:p}`agha1986actors`) and each link corresponds to data passed between actors. For instance, a simple experiment in which calcium fluorescence images are read in from a microscope, processed via [CaImAn](https://github.com/flatironinstitute/CaImAn), fit using a [linear-nonlinear-Poisson model](https://en.wikipedia.org/wiki/Linear-nonlinear-Poisson_cascade_model) along with information about the stimulus presented, and displayed via a GUI might look something like + +:::{figure-md} example_dag + + +An example directed graph corresponding to an _improv_ experimental pipeline. +::: + +Pipelines in _improv_ are specified by [YAML files](https://yaml.org). _improv_ configuration files contain three top-level headings: +1. `settings` includes program settings to be passed to `nexus.Nexus.createNexus` upon startup. This includes control and output port numbers to be used for input and output to the server, respectively. See the documentation for `nexus.Nexus.createNexus` for other arguments. +1. `actors` is a list of actors that form the nodes of a directed graph. Each item requires two attributes: `package` gives the name of the Python file containing the actor definition and `class` gives the name of the class within that file. As described in [](page:running:options), _improv_ will search for actors in the directory containing the YAML config file by default, though more directories can be specified with the `--actor-path` option. **Other attributes will be passed directly (as a dictionary) to the actor class constructor.** +1. `connections` is a list of connections between actors. Each item contains the name of an actor output (e.g., `Processor.q_out`) and a list of actors that will receive this output. + +The example graph of [the figure above](example_dag) is implemented in the [zebrafish demo](https://github.com/project-improv/improv/blob/main/demos/naumann/naumann_demo.yaml), whose YAML file is given by +``` +actors: + GUI: + package: actors.visual_model + class: DisplayVisual + visual: Visual + + Acquirer: + package: improv.actors.acquire + class: TiffAcquirer + filename: data/recent/xx.tif + framerate: 2 + + Processor: + package: actors.processor + class: CaimanProcessor + init_filename: data/recent/xx.tif + config_file: naumann_caiman_params.txt + + Visual: + package: actors.visual_model + class: CaimanVisual + + Analysis: + package: actors.analysis_model + class: ModelAnalysis + + InputStim: + package: improv.actors.acquire + class: StimAcquirer + filename: data/recent/stim_freq.txt + +connections: + Acquirer.q_out: [Processor.q_in, Visual.raw_frame_queue] + Processor.q_out: [Analysis.q_in] + Analysis.q_out: [Visual.q_in] + InputStim.q_out: [Analysis.input_stim_queue] +``` +The particular details are not so important here as the logic of how pieces of the experiment are put together into a single file. + +## Interprocess communication + +To understand how _improv_ translates a pipeline specified in a YAML file to a working experiment, it's helpful to consider what happens after `improv run` is called: +1. An instance of the server is created using the specified configuration file and ports. + 1. The configuration file is loaded and parsed. Ports specified in the configuration file are overridden by ports specified at the command line. + 1. If no ports were specified, random available ports are chosen. One port (`control_port`) is for incoming instructions to the server (e.g., from GUI, TUI, etc.). The other port (`output_port`) is for broadcast status messages from the server. + 1. The server starts the in-memory data store (with size specified (in bytes) in the `settings` section of the YAML file). + 1. The server connects to the store and subscribes to its notifications. + 1. The server loops over actors in the configuration file, creating an instance of each class for each actor. + 1. The server loops over connections, creating a communication channel between each pair of actors. +1. The server is started. + 1. Using [`multiprocessing`](https://docs.python.org/3/library/multiprocessing.html), each actor's `run` method is launched (via either spawn or fork, as specified by the actor's `method` attribute in the YAML file) in a separate process.[^run_warning] + 1. The server starts an event loop that listens for input from either the control port or the actors. An "Awaiting input" message is sent on the output port. + 1. The server writes its port configuration to the log file, to be read by clients who wish to connect. +1. The textual user interface (TUI) client is started and connects to the server. Other clients may also connect to the server's control port and send commands. +1. At this point, clients may send any of the messages defined in `improv.actor.Signal`, including `setup`, `run`, `stop`, and `quit`. See [](page:signals) for more details. + +What is also important to realize is that none of the above directly pertains to data flow. Once the links between actors are set up, each actor's `run` method is responsible for +1. listening on its incoming links for addresses (keys) of newly available data from each parent actor in the graph +1. retrieving new data items directly from the store (by key) +1. performing whatever processing is required +1. depositing its outputs in the store +1. broadcasting the address(es) of its data outputs to its children in the graph + +For examples and further documentation, see [](page:actors). + +[^run_warning]: Note that users will _not_, in most cases, overload the `run` command for actors. They should instead write the `runStep` function. See [](page:actors). + +## Logging and persistence +Finally, _improv_ handles centralized logging via the [`logging`](https://docs.python.org/3/library/logging.html) module, which listens for messages on a global logging port. These messages are written to the experimental log file. + +Data from the server are persisted to disk using [LMDB](http://www.lmdb.tech/doc/) (if `settings: use_hdd` is set to `true` in the configuration file). \ No newline at end of file diff --git a/docs/installation.md b/docs/installation.md new file mode 100644 index 000000000..6fa470ddd --- /dev/null +++ b/docs/installation.md @@ -0,0 +1,112 @@ +(page:installation)= +# Installation and building + +## Simple installation +The simplest way to install _improv_ is with pip: +``` +pip install improv +``` +````{warning} +Due to [this pyzmq issue](https://github.com/zeromq/libzmq/issues/3313), if you're running on Ubuntu, you need to specify +``` +pip install improv --no-binary pyzmq +``` +to build `pyzmq` from source. +```` + +## Required dependencies + +### Redis + +_improv_ uses Redis, an in-memory datastore, to hold data to be communicated between actors. _improv_ has been tested with Redis server version 7.2.4. Please refer to the instructions below for your operating system: + +#### macOS +A compatible version of Redis can be installed via Homebrew: +``` +brew install redis +``` + +#### Linux +A compatible version of Redis can be installed for most standard Linux distributions by following Redis' instructions on their [installation guide](https://redis.io/docs/latest/operate/oss_and_stack/install/install-redis/install-redis-on-linux/). + +#### Windows (WSL2) +Redis can also be installed on Windows in WSL2. The [installation guide](https://redis.io/docs/latest/operate/oss_and_stack/install/install-redis/install-redis-on-windows/) details the process for both the Windows and Linux portions of WSL2. + +## Optional dependencies +In addition to the basic _improv_ installation, users who want to, e.g., run tests locally and build docs should do +``` +pip install improv[tests,docs] +``` +(on bash) or +``` +pip install "improv[tests,docs]" +``` +(on zsh for newer Macs). + +(page:installation:source_build)= +## Building from source +Users who want to run the demos or contribute to the project will want to build and install from source. + +You can either obtain the source code from our [current releases](https://github.com/project-improv/improv/releases) (download and unzip the `Source code (zip)` file) or via cloning from GitHub: +```bash +git clone https://github.com/project-improv/improv.git +``` + +With the code downloaded, you'll then need to build and install the package. For installation, we recommend using a virtual environment so that _improv_ and its dependencies don't conflict with your existing Python setup. For instance, using [`mamba`](https://mamba.readthedocs.io/en/latest/): +```bash +mamba create -n improv python=3.10 +``` +will create an `improv` environment running Python 3.10 that you can activate via +```bash +mamba activate improv +``` +Other options, such as [`virtualenv`](https://virtualenv.pypa.io/en/latest/) will also work. ([`conda`](https://docs.conda.io/projects/conda/en/stable/) is largely superseded by `mamba`.) + +Currently, we build using [Setuptools](https://setuptools.pypa.io/en/latest/index.html) via a `pyproject.toml` file as specified in [PEP 518](https://peps.python.org/pep-0518/). This may allow us to switch out the build backend (or frontend) later. + +For now, the package can be built by running +``` +python -m build +``` +from within the project directory. After that +``` +pip install --editable . +``` +will install the package in editable mode, which means that changes in the project directory will affect the code that is run (i.e., the installation will not copy over the code to `site-packages` but simply link the project directory). + +When uninstalling, be sure to do so _from outside the project directory_, since otherwise, `pip` only appears to find the command line script, not the full package. + + +## Building documents locally +Currently, we have [GitHub Actions](https://docs.github.com/en/actions) set up to build the Jupyter Book [automatically](https://jupyterbook.org/en/stable/publish/gh-pages.html#automatically-host-your-book-with-github-actions). But for previewing changes locally, you will want to build the documents and check them yourself. + +To do that, you'll first need to install the docs dependencies with +``` +pip install improv[docs] +``` +or +``` +pip install -e .[docs] +``` +Then simply run +``` +jupyter-book build docs +``` +and open `docs/_build/html/index.html` in your browser. + +## Getting around certificate issues + +On some systems, building (or installing from `pip`) can run into [this error](https://stackoverflow.com/questions/25981703/pip-install-fails-with-connection-error-ssl-certificate-verify-failed-certi) related to SSL certificates. For `pip`, the solution is given in the linked StackOverflow question (add `--trusted-host` to the command line), but for `build`, we run into the issue that the `trusted-host` flag will not be passed through to `pip`, and `pip` will build inside an isolated venv, meaning it won't read a file-based configuration option like the one given in the answer, either. + +The (inelegant) solution that will work is to set the `pip.conf` file with +``` +[global] +trusted-host = pypi.python.org + pypi.org + files.pythonhosted.org +``` +and then run +``` +python -m build --no-isolation +``` +which will allow `pip` to correctly read the configuration. \ No newline at end of file diff --git a/docs/intro.md b/docs/intro.md index 95577c402..eb8d7aa07 100644 --- a/docs/intro.md +++ b/docs/intro.md @@ -1,11 +1,33 @@ -# Welcome to your Jupyter Book +# _improv_: A platform for adaptive neuroscience experiments -This is a small sample book to give you a feel for how book content is -structured. -It shows off a few of the major file types, as well as some sample content. -It does not go in-depth into any particular topic - check out [the Jupyter Book documentation](https://jupyterbook.org) for more information. -Check out the content pages bundled with this sample book to see more. +_improv_ is a lightweight package designed to enable complex adaptive experiments in neuroscience. This includes both traditional closed-loop experiments, in which low latency and real-time analysis are the priority, and experiments that attempt to maximize neural responses by changing the stimulus presented. In each case, the design of the experiment must change online based on previously seen data. -```{tableofcontents} -``` \ No newline at end of file +In practice, such experiments are technically challenging, since they require coordinating hardware with software, data acquisition and preprocessing with analysis. We build _improv_ to provide a simple means of tackling these coordination problems. In _improv_, you specify your experiment as a data flow graph, write simple Python classes for each step, and we handle the rest. + +:::{figure-md} improv_gif + + +Raw two-photon calcium imaging data in zebrafish (left), with cells detected in real time by [CaImAn](https://github.com/flatironinstitute/CaImAn) (right). Neurons have been colored by directional tuning curves and functional connectivity (lines) estimated online, during a live experiment. Here only a few minutes of data have been acquired, and neurons are colored by their strongest response to visual simuli shown so far. +::: + +## Why _improv_? + +### Skip the messy parts +Need to collect data from multiple sources? Run processes on different machines? Pipeline data to multiple downstream analyses? _improv_ handles the details of communication between processes and machines so you can focus on what's important: defining the logic of your experiment. + + + +### Simple design +_improv_'s design is based on a simplified version of the Actor Model of concurrency {cite:p}`agha1986actors`: the experiment is a directed graph, the nodes are actors, and data flows by asynchronous message passing. Apart from this, we strive for maximum flexibility. _improv_ allows for arbitrary Python code in actors, allowing you to interoperate with the widest variety of tools. Examples can be found in [](page:demos) and our [`improv-sketches`](https://github.com/project-improv/improv-sketches) repository. + + + +### Speed +_improv_ is designed to be fast enough for real-time experiments that need millisecond-scale latencies. By minimizing data passing and making use of in-memory data stores, we find that many applications are limited by network latency, _not_ processing time. + +### Reliability +_improv_ is designed from the ground up to be fault-tolerant, with a priority on data integrity. Individual processes may crash, but the system will keep running, ensuring that a single bad line of code does not result in data loss. In addition, we provide the tools to produce an audit trail of everything that happened online, ensuring you don't forget that one setting you changed mid-experiment. + + \ No newline at end of file diff --git a/docs/references.bib b/docs/references.bib index 87e609880..bed09259f 100644 --- a/docs/references.bib +++ b/docs/references.bib @@ -1,55 +1,9 @@ --- --- -@inproceedings{holdgraf_evidence_2014, - address = {Brisbane, Australia, Australia}, - title = {Evidence for {Predictive} {Coding} in {Human} {Auditory} {Cortex}}, - booktitle = {International {Conference} on {Cognitive} {Neuroscience}}, - publisher = {Frontiers in Neuroscience}, - author = {Holdgraf, Christopher Ramsay and de Heer, Wendy and Pasley, Brian N. and Knight, Robert T.}, - year = {2014} -} - -@article{holdgraf_rapid_2016, - title = {Rapid tuning shifts in human auditory cortex enhance speech intelligibility}, - volume = {7}, - issn = {2041-1723}, - url = {http://www.nature.com/doifinder/10.1038/ncomms13654}, - doi = {10.1038/ncomms13654}, - number = {May}, - journal = {Nature Communications}, - author = {Holdgraf, Christopher Ramsay and de Heer, Wendy and Pasley, Brian N. and Rieger, Jochem W. and Crone, Nathan and Lin, Jack J. and Knight, Robert T. and Theunissen, Frédéric E.}, - year = {2016}, - pages = {13654}, - file = {Holdgraf et al. - 2016 - Rapid tuning shifts in human auditory cortex enhance speech intelligibility.pdf:C\:\\Users\\chold\\Zotero\\storage\\MDQP3JWE\\Holdgraf et al. - 2016 - Rapid tuning shifts in human auditory cortex enhance speech intelligibility.pdf:application/pdf} -} - -@inproceedings{holdgraf_portable_2017, - title = {Portable learning environments for hands-on computational instruction using container-and cloud-based technology to teach data science}, - volume = {Part F1287}, - isbn = {978-1-4503-5272-7}, - doi = {10.1145/3093338.3093370}, - abstract = {© 2017 ACM. There is an increasing interest in learning outside of the traditional classroom setting. This is especially true for topics covering computational tools and data science, as both are challenging to incorporate in the standard curriculum. These atypical learning environments offer new opportunities for teaching, particularly when it comes to combining conceptual knowledge with hands-on experience/expertise with methods and skills. Advances in cloud computing and containerized environments provide an attractive opportunity to improve the effciency and ease with which students can learn. This manuscript details recent advances towards using commonly-Available cloud computing services and advanced cyberinfrastructure support for improving the learning experience in bootcamp-style events. We cover the benets (and challenges) of using a server hosted remotely instead of relying on student laptops, discuss the technology that was used in order to make this possible, and give suggestions for how others could implement and improve upon this model for pedagogy and reproducibility.}, - author = {Holdgraf, Christopher Ramsay and Culich, A. and Rokem, A. and Deniz, F. and Alegro, M. and Ushizima, D.}, - year = {2017}, - keywords = {Teaching, Bootcamps, Cloud computing, Data science, Docker, Pedagogy} -} - -@article{holdgraf_encoding_2017, - title = {Encoding and decoding models in cognitive electrophysiology}, - volume = {11}, - issn = {16625137}, - doi = {10.3389/fnsys.2017.00061}, - abstract = {© 2017 Holdgraf, Rieger, Micheli, Martin, Knight and Theunissen. Cognitive neuroscience has seen rapid growth in the size and complexity of data recorded from the human brain as well as in the computational tools available to analyze this data. This data explosion has resulted in an increased use of multivariate, model-based methods for asking neuroscience questions, allowing scientists to investigate multiple hypotheses with a single dataset, to use complex, time-varying stimuli, and to study the human brain under more naturalistic conditions. These tools come in the form of “Encoding” models, in which stimulus features are used to model brain activity, and “Decoding” models, in which neural features are used to generated a stimulus output. Here we review the current state of encoding and decoding models in cognitive electrophysiology and provide a practical guide toward conducting experiments and analyses in this emerging field. Our examples focus on using linear models in the study of human language and audition. We show how to calculate auditory receptive fields from natural sounds as well as how to decode neural recordings to predict speech. The paper aims to be a useful tutorial to these approaches, and a practical introduction to using machine learning and applied statistics to build models of neural activity. The data analytic approaches we discuss may also be applied to other sensory modalities, motor systems, and cognitive systems, and we cover some examples in these areas. In addition, a collection of Jupyter notebooks is publicly available as a complement to the material covered in this paper, providing code examples and tutorials for predictive modeling in python. The aimis to provide a practical understanding of predictivemodeling of human brain data and to propose best-practices in conducting these analyses.}, - journal = {Frontiers in Systems Neuroscience}, - author = {Holdgraf, Christopher Ramsay and Rieger, J.W. and Micheli, C. and Martin, S. and Knight, R.T. and Theunissen, F.E.}, - year = {2017}, - keywords = {Decoding models, Encoding models, Electrocorticography (ECoG), Electrophysiology/evoked potentials, Machine learning applied to neuroscience, Natural stimuli, Predictive modeling, Tutorials} -} - -@book{ruby, - title = {The Ruby Programming Language}, - author = {Flanagan, David and Matsumoto, Yukihiro}, - year = {2008}, - publisher = {O'Reilly Media} +@book{agha1986actors, + title={Actors: a model of concurrent computation in distributed systems}, + author={Agha, Gul}, + year={1986}, + publisher={MIT press} } \ No newline at end of file diff --git a/docs/running.ipynb b/docs/running.ipynb new file mode 100644 index 000000000..47b7b2b08 --- /dev/null +++ b/docs/running.ipynb @@ -0,0 +1,227 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "(page:running)=\n", + "# Running _improv_\n", + "\n", + "## Basic invocation\n", + "While _improv_ can be invoked via API, the simplest method is to run it from the `improv` command line interface:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": { + "vscode": { + "languageId": "shellscript" + } + }, + "outputs": [], + "source": [ + "!improv --help" + ] + }, + { + "cell_type": "markdown", + "metadata": { + "vscode": { + "languageId": "shellscript" + } + }, + "source": [ + "The most basic command is `improv run`:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": { + "vscode": { + "languageId": "shellscript" + } + }, + "outputs": [], + "source": [ + "!improv run --help" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "For instance, if you run the minimal demo\n", + "```bash\n", + "improv run demos/minimal/minimal.yaml\n", + "```\n", + "_improv_ will use the YAML file to setup up and run both the _improv_ server and the _improv_ client, which is a text user interface (TUI):" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "
\n",
+ "\n", + "
" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "In the text window, we can issue commands to the _improv_ server:\n", + "- **setup:** to initialize all actors and create their connections\n", + "- **run:** to start the experiment\n", + "- **stop:** to send the stop signal to all actors\n", + "- **quit:** to initiate cleanup and exit the TUI.\n", + "\n", + "More details can be found at [](page:signals)." + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "(page:running:options)=\n", + "## Command line options\n", + "For more advanced usage, `improv run` lets you specify the ports to use for communicating with the server and client (the default is to use random available ports), as well as specifying the name of the log file. The default is to _append_ to `global.log`, so be sure to either (a) delete this file periodically (not recommended) or (b) use a unique log file name for each experiment.\n", + "\n", + "```{tip}\n", + "A particularly important command line option is `--actor-path` or `-a`, which gives a list of directories in which to search for the Python modules containing actors. The default actor path is the directory containing the configuration file.\n", + "```" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Other subcommands\n", + "In addition, for running _improv_ across multiple machines, there is the `improv server` command" + ] + }, + { + "cell_type": "code", + "execution_count": 1, + "metadata": { + "vscode": { + "languageId": "shellscript" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "usage: improv server [-h] [-c CONTROL_PORT] [-o OUTPUT_PORT] [-l LOGGING_PORT]\n", + " [-f LOGFILE] [-a ACTOR_PATH]\n", + " configfile\n", + "\n", + "Start the improv server\n", + "\n", + "positional arguments:\n", + " configfile YAML file specifying improv pipeline\n", + "\n", + "options:\n", + " -h, --help show this help message and exit\n", + " -c CONTROL_PORT, --control-port CONTROL_PORT\n", + " local port on which control signals are received\n", + " -o OUTPUT_PORT, --output-port OUTPUT_PORT\n", + " local port on which output messages are broadcast\n", + " -l LOGGING_PORT, --logging-port LOGGING_PORT\n", + " local port on which logging messages are broadcast\n", + " -f LOGFILE, --logfile LOGFILE\n", + " name of log file\n", + " -a ACTOR_PATH, --actor-path ACTOR_PATH\n", + " search path to add to sys.path when looking for\n", + " actors; defaults to the directory containing\n", + " configfile\n" + ] + } + ], + "source": [ + "!improv server --help" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "to start the server and `improv client` to start a client locally:" + ] + }, + { + "cell_type": "code", + "execution_count": 2, + "metadata": { + "vscode": { + "languageId": "shellscript" + } + }, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "usage: improv client [-h] [-c CONTROL_PORT] [-s SERVER_PORT] [-l LOGGING_PORT]\n", + "\n", + "Start the improv client\n", + "\n", + "options:\n", + " -h, --help show this help message and exit\n", + " -c CONTROL_PORT, --control-port CONTROL_PORT\n", + " address on which control signals are sent to the\n", + " server\n", + " -s SERVER_PORT, --server-port SERVER_PORT\n", + " address on which messages from the server are received\n", + " -l LOGGING_PORT, --logging-port LOGGING_PORT\n", + " address on which logging messages are broadcast\n" + ] + } + ], + "source": [ + "!improv client --help" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "```{note}\n", + "When running clients and servers on different machines, you _will_ need to know and specify the relevant ports, so it may be good to select these manually.\n", + "```" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Cleaning up\n", + "Finally, it occasionally happens that processes do not shut down cleanly. This can leave orphaned processes running on the system, which quickly eats memory. To see these processes, you can run `improv list`, which will show all processes currently associated with _improv_. Running `improv cleanup` will prompt before killing all these processes." + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "improv", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.10.11" + } + }, + "nbformat": 4, + "nbformat_minor": 2 +} diff --git a/docs/signals.md b/docs/signals.md new file mode 100644 index 000000000..345b61bb3 --- /dev/null +++ b/docs/signals.md @@ -0,0 +1,40 @@ +(page:signals)= +# Signals and communicating between actors +In [](page:actors), we covered the basics of writing your own `Actor` class. Here, we consider a further complication: +How do actors talk with _improv_ and with one another. The most basic mechanism for doing so is through _signals_. + +On Unix-like computing systems, signals are messages sent by the operating system to processes running on the machine. +The most well-known of these are `SIGINT` ("interrupt," what you get by pressing `Ctrl-C` at the terminal), +`SIGTERM` ("terminate," shut the process down, but ask nicely), and `SIGKILL` ("kill," and don't take no for an +answer). In Python, handling of these events is done with the [`signal`](https://docs.python.org/3/library/signal.html) +library. + +For communicating between _improv_'s server and actors, we use both the `signal` library (internally) and +`actor.Signal` for _improv_'s own set of messages. In the default `Actor` class (an alias for `ManagedActor`), +signals are handled by the `RunManger` class, which calls the relevant class method when it receives a signal. +In [](tables:signals) we list the signals defined in `Signal` along with the `Actor` methods they call. + +```{table} Correspondences between signals +:name: tables:signals + +| `actor.Signal` received | `ManagedActor` method called | +|---|---| +| `setup` | `setup` | +| `run` | `runStep` | +| `pause` | not yet implemented | +| `resume` | not yet implemented | +| `reset` | not yet implemented | +| `load` | not yet implemented | +| `ready` | sent by actors to server | +| `kill` | handled by server | +| `revive` | not yet implemented | +| `stop` | `stop` | +| `stop_success` | not yet implemented | +| `quit` | handled by server | +``` + +```{note} +Users who wish to do their own signal handling (e.g., by inheriting from `AbstractActor`) will need to test for +the presence of `actor.Signal` messages in the connection to the server within the `run` function and handle +them appropriately. +``` \ No newline at end of file diff --git a/env/improv-suite2p_env.yml b/env/improv-suite2p_env.yml index 23895f770..bca783921 100644 --- a/env/improv-suite2p_env.yml +++ b/env/improv-suite2p_env.yml @@ -38,5 +38,4 @@ dependencies: - natsort - scanimage-tiff-reader - rastermap==0.1.0 - - lmdb - suite2p==0.6.16 diff --git a/env/improv.yml b/env/improv.yml index 5978d7212..6891cd8e5 100644 --- a/env/improv.yml +++ b/env/improv.yml @@ -8,5 +8,4 @@ dependencies: - pip - pyarrow - pyqt -- python-lmdb - pyyaml \ No newline at end of file diff --git a/env/improv_env.yml b/env/improv_env.yml index c08a93b79..389cb616a 100644 --- a/env/improv_env.yml +++ b/env/improv_env.yml @@ -225,6 +225,6 @@ dependencies: - pip: - argh - line-profiler - - lmdb==0.97 + - redis - pathtools - watchdog diff --git a/improv/actor.py b/improv/actor.py index 10d0baf74..3d6a2a237 100644 --- a/improv/actor.py +++ b/improv/actor.py @@ -3,6 +3,8 @@ import asyncio import traceback from queue import Empty + +import improv.store from improv.store import StoreInterface import logging @@ -18,7 +20,9 @@ class AbstractActor: Also needs to be responsive to sent Signals (e.g. run, setup, etc) """ - def __init__(self, name, store_loc, method="fork"): + def __init__( + self, name, store_loc=None, method="fork", store_port_num=None, *args, **kwargs + ): """Require a name for multiple instances of the same actor/class Create initial empty dict of Links for easier referencing """ @@ -29,6 +33,7 @@ def __init__(self, name, store_loc, method="fork"): self.client = None self.store_loc = store_loc self.lower_priority = False + self.store_port_num = store_port_num # Start with no explicit data queues. # q_in and q_out are reserved for passing ID information @@ -55,7 +60,11 @@ def setStoreInterface(self, client): def _getStoreInterface(self): # TODO: Where do we require this be run? Add a Signal and include in RM? if not self.client: - store = StoreInterface(self.name, self.store_loc) + store = None + if StoreInterface == improv.store.RedisStoreInterface: + store = StoreInterface(self.name, self.store_port_num) + else: + store = StoreInterface(self.name, self.store_loc) self.setStoreInterface(store) def setLinks(self, links): @@ -182,7 +191,7 @@ def changePriority(self): class ManagedActor(AbstractActor): def __init__(self, *args, **kwargs): - super().__init__(*args) + super().__init__(*args, **kwargs) # Define dictionary of actions for the RunManager self.actions = {} diff --git a/improv/cli.py b/improv/cli.py index 75c00aac7..720d4e2f8 100644 --- a/improv/cli.py +++ b/improv/cli.py @@ -2,6 +2,7 @@ import os.path import re import argparse +import signal import subprocess import sys import psutil @@ -253,12 +254,12 @@ def run_server(args): def run_list(args, printit=True): out_list = [] - pattern = re.compile(r"(improv (run|client|server)|plasma_store)") - mp_pattern = re.compile(r"-c from multiprocessing") + pattern = re.compile(r"(improv (run|client|server)|plasma_store|redis-server)") + # mp_pattern = re.compile(r"-c from multiprocessing") # TODO is this right? for proc in psutil.process_iter(["pid", "name", "cmdline"]): if proc.info["cmdline"]: cmdline = " ".join(proc.info["cmdline"]) - if re.search(pattern, cmdline) or re.search(mp_pattern, cmdline): + if re.search(pattern, cmdline): # or re.search(mp_pattern, cmdline): out_list.append(proc) if printit: print(f"{proc.pid} {proc.name()} {cmdline}") @@ -280,14 +281,23 @@ def run_cleanup(args, headless=False): if res.lower() == "y": for proc in proc_list: - if not proc.status == "terminated": + if not proc.status() == psutil.STATUS_STOPPED: + logging.info( + f"process {proc.pid} {proc.name()}" + f" has status {proc.status()}. Interrupting." + ) try: - proc.terminate() + proc.send_signal(signal.SIGINT) except psutil.NoSuchProcess: pass gone, alive = psutil.wait_procs(proc_list, timeout=3) for p in alive: - p.kill() + p.send_signal(signal.SIGINT) + try: + p.wait(timeout=10) + except psutil.TimeoutExpired as e: + logging.warning(f"{e}: Process did not exit on time.") + else: if not headless: print("No running processes found.") diff --git a/improv/config.py b/improv/config.py index 00f52382d..5f73c18b6 100644 --- a/improv/config.py +++ b/improv/config.py @@ -108,6 +108,10 @@ def createConfig(self): raise RepeatedConnectionsError(name) self.connections.update({name: conn}) + + if "datastore" in cfg.keys(): + self.datastore = cfg["datastore"] + return 0 def addParams(self, type, param): @@ -125,6 +129,60 @@ def saveActors(self): for a in self.actors.values(): wflag = a.saveConfigModules(pathName, wflag) + def use_plasma(self): + return "plasma_config" in self.config.keys() + + def get_redis_port(self): + if self.redis_port_specified(): + return self.config["redis_config"]["port"] + else: + return Config.get_default_redis_port() + + def redis_port_specified(self): + if "redis_config" in self.config.keys(): + return "port" in self.config["redis_config"] + return False + + def redis_saving_enabled(self): + if "redis_config" in self.config.keys(): + return ( + self.config["redis_config"]["enable_saving"] + if "enable_saving" in self.config["redis_config"] + else None + ) + + def generate_ephemeral_aof_dirname(self): + if "redis_config" in self.config.keys(): + return ( + self.config["redis_config"]["generate_ephemeral_aof_dirname"] + if "generate_ephemeral_aof_dirname" in self.config["redis_config"] + else None + ) + return False + + def get_redis_aof_dirname(self): + if "redis_config" in self.config.keys(): + return ( + self.config["redis_config"]["aof_dirname"] + if "aof_dirname" in self.config["redis_config"] + else None + ) + return None + + def get_redis_fsync_frequency(self): + if "redis_config" in self.config.keys(): + frequency = ( + self.config["redis_config"]["fsync_frequency"] + if "fsync_frequency" in self.config["redis_config"] + else None + ) + + return frequency + + @staticmethod + def get_default_redis_port(): + return "6379" + class ConfigModule: def __init__(self, name, packagename, classname, options=None): diff --git a/improv/nexus.py b/improv/nexus.py index a750c3447..b5fe19c4e 100644 --- a/improv/nexus.py +++ b/improv/nexus.py @@ -1,4 +1,5 @@ import os +import time import uuid import signal import logging @@ -14,7 +15,7 @@ import zmq.asyncio as zmq from zmq import PUB, REP, SocketOption -from improv.store import StoreInterface +from improv.store import StoreInterface, RedisStoreInterface, PlasmaStoreInterface from improv.actor import Signal from improv.config import Config from improv.link import Link, MultiLink @@ -29,7 +30,12 @@ class Nexus: """Main server class for handling objects in improv""" def __init__(self, name="Server"): + self.redis_fsync_frequency = None + self.store = None + self.config = None self.name = name + self.aof_dir = None + self.redis_saving_enabled = False def __str__(self): return self.name @@ -37,7 +43,6 @@ def __str__(self): def createNexus( self, file=None, - use_hdd=False, use_watcher=None, store_size=10_000_000, control_port=0, @@ -51,7 +56,6 @@ def createNexus( Args: file (string): Name of the config file. - use_hdd (bool): Whether to use hdd for the store. use_watcher (bool): Whether to use watcher for the store. store_size (int): initial store size control_port (int): port number for input socket @@ -76,8 +80,6 @@ def createNexus( # set config options loaded from file # in Python 3.9, can just merge dictionaries using precedence cfg = self.config.settings - if "use_hdd" not in cfg: - cfg["use_hdd"] = use_hdd if "use_watcher" not in cfg: cfg["use_watcher"] = use_watcher if "store_size" not in cfg: @@ -89,6 +91,7 @@ def createNexus( # set up socket in lieu of printing to stdout self.zmq_context = zmq.Context() + self.zmq_context.setsockopt(SocketOption.LINGER, 1) self.out_socket = self.zmq_context.socket(PUB) self.out_socket.bind("tcp://*:%s" % cfg["output_port"]) out_port_string = self.out_socket.getsockopt_string(SocketOption.LAST_ENDPOINT) @@ -99,20 +102,26 @@ def createNexus( in_port_string = self.in_socket.getsockopt_string(SocketOption.LAST_ENDPOINT) cfg["control_port"] = int(in_port_string.split(":")[-1]) + self.configure_redis_persistence() + # default size should be system-dependent - self._startStoreInterface(store_size) + if self.config and self.config.use_plasma(): + self._startStoreInterface(store_size) + else: + self._startStoreInterface(store_size) + logger.info("Redis server started") + self.out_socket.send_string("StoreInterface started") # connect to store and subscribe to notifications logger.info("Create new store object") - self.store = StoreInterface(store_loc=self.store_loc) - self.store.subscribe() + if self.config and self.config.use_plasma(): + self.store = PlasmaStoreInterface(store_loc=self.store_loc) + else: + self.store = StoreInterface(server_port_num=self.store_port) + logger.info(f"Redis server connected on port {self.store_port}") - # LMDB storage - self.use_hdd = cfg["use_hdd"] - if self.use_hdd: - self.lmdb_name = f'lmdb_{datetime.now().strftime("%Y%m%d_%H%M%S")}' - self.store_dict = dict() + self.store.subscribe() # TODO: Better logic/flow for using watcher as an option self.p_watch = None @@ -227,6 +236,77 @@ def initConfig(self): watchin.append(watch_link) self.createWatcher(watchin) + def configure_redis_persistence(self): + # invalid configs: specifying filename and using an ephemeral filename, + # specifying that saving is off but providing either filename option + aof_dirname = self.config.get_redis_aof_dirname() + generate_unique_dirname = self.config.generate_ephemeral_aof_dirname() + redis_saving_enabled = self.config.redis_saving_enabled() + redis_fsync_frequency = self.config.get_redis_fsync_frequency() + + if aof_dirname and generate_unique_dirname: + logger.error( + "Cannot both generate a unique dirname and use the one provided." + ) + raise Exception("Cannot use unique dirname and use the one provided.") + + if aof_dirname or generate_unique_dirname or redis_fsync_frequency: + if redis_saving_enabled is None: + redis_saving_enabled = True + elif not redis_saving_enabled: + logger.error( + "Invalid configuration. Cannot save to disk with saving disabled." + ) + raise Exception("Cannot persist to disk with saving disabled.") + + self.redis_saving_enabled = redis_saving_enabled + + if redis_fsync_frequency and redis_fsync_frequency not in [ + "every_write", + "every_second", + "no_schedule", + ]: + logger.error("Cannot use unknown fsync frequency ", redis_fsync_frequency) + raise Exception( + "Cannot use unknown fsync frequency ", redis_fsync_frequency + ) + + if redis_fsync_frequency is None: + redis_fsync_frequency = "no_schedule" + + if redis_fsync_frequency == "every_write": + self.redis_fsync_frequency = "always" + elif redis_fsync_frequency == "every_second": + self.redis_fsync_frequency = "everysec" + elif redis_fsync_frequency == "no_schedule": + self.redis_fsync_frequency = "no" + else: + logger.error("Unknown fsync frequency ", redis_fsync_frequency) + raise Exception("Unknown fsync frequency ", redis_fsync_frequency) + + if aof_dirname: + self.aof_dir = aof_dirname + elif generate_unique_dirname: + self.aof_dir = "improv_persistence_" + str(uuid.uuid1()) + + if self.redis_saving_enabled and self.aof_dir is not None: + logger.info( + "Redis saving enabled. Saving to directory " + + self.aof_dir + + " on schedule " + + "'{}'".format(self.redis_fsync_frequency) + ) + elif self.redis_saving_enabled: + logger.info( + "Redis saving enabled with default directory " + + "on schedule " + + "'{}'".format(self.redis_fsync_frequency) + ) + else: + logger.info("Redis saving disabled.") + + return + def startNexus(self): """ Puts all actors in separate processes and begins polling @@ -299,6 +379,13 @@ def destroyNexus(self): ) logger.warning("Delete the store at location {0}".format(self.store_loc)) + if hasattr(self, "out_socket"): + self.out_socket.close(linger=0) + if hasattr(self, "in_socket"): + self.in_socket.close(linger=0) + if hasattr(self, "zmq_context"): + self.zmq_context.destroy(linger=0) + async def pollQueues(self): """ Listens to links and processes their signals. @@ -573,18 +660,13 @@ def stop_polling(self, stop_signal, queues): logger.info("Polling has stopped.") def createStoreInterface(self, name): - """Creates StoreInterface w/ or w/out LMDB - functionality based on {self.use_hdd}.""" - if not self.use_hdd: - return StoreInterface(name, self.store_loc) + """Creates StoreInterface""" + if self.config.use_plasma(): + return PlasmaStoreInterface(name, self.store_loc) else: - if name not in self.store_dict: - self.store_dict[name] = StoreInterface( - name, self.store_loc, use_hdd=True, lmdb_name=self.lmdb_name - ) - return self.store_dict[name] + return RedisStoreInterface(server_port_num=self.store_port) - def _startStoreInterface(self, size): + def _startStoreInterface(self, size, attempts=20): """Start a subprocess that runs the plasma store Raises a RuntimeError exception size is undefined Raises an Exception if the plasma store doesn't start @@ -596,12 +678,14 @@ def _startStoreInterface(self, size): Raises: RuntimeError: if the size is undefined - Exception: if the plasma store doesn't start + Exception: if the store doesn't start """ if size is None: raise RuntimeError("Server size needs to be specified") - try: + self.use_plasma = False + if self.config and self.config.use_plasma(): + self.use_plasma = True self.store_loc = str(os.path.join("/tmp/", str(uuid.uuid4()))) self.p_StoreInterface = subprocess.Popen( [ @@ -617,19 +701,111 @@ def _startStoreInterface(self, size): stderr=subprocess.DEVNULL, ) logger.info("StoreInterface start successful: {}".format(self.store_loc)) - except Exception as e: - logger.exception("StoreInterface cannot be started: {}".format(e)) + else: + logger.info("Setting up Redis store.") + self.store_port = ( + self.config.get_redis_port() + if self.config and self.config.redis_port_specified() + else Config.get_default_redis_port() + ) + if self.config and self.config.redis_port_specified(): + logger.info( + "Attempting to connect to Redis on port {}".format(self.store_port) + ) + # try with failure, incrementing port number + self.p_StoreInterface = self.start_redis(size) + time.sleep(3) + if self.p_StoreInterface.poll(): + logger.error("Could not start Redis on specified port number.") + raise Exception("Could not start Redis on specified port.") + else: + logger.info("Redis port not specified. Searching for open port.") + for attempt in range(attempts): + logger.info( + "Attempting to connect to Redis on port {}".format( + self.store_port + ) + ) + # try with failure, incrementing port number + self.p_StoreInterface = self.start_redis(size) + time.sleep(3) + if self.p_StoreInterface.poll(): # Redis could not start + logger.info( + "Could not connect to port {}".format(self.store_port) + ) + self.store_port = str(int(self.store_port) + 1) + else: + break + else: + logger.error("Could not start Redis on any tried port.") + raise Exception("Could not start Redis on any tried ports.") + + logger.info(f"StoreInterface start successful on port {self.store_port}") + + def start_redis(self, size): + subprocess_command = [ + "redis-server", + "--port", + str(self.store_port), + "--maxmemory", + str(size), + "--save", # this only turns off RDB, which we want permanently off + '""', + ] + + if self.aof_dir is not None and len(self.aof_dir) == 0: + raise Exception("Persistence directory specified but no filename given.") + + if self.aof_dir is not None: # use specified (possibly pre-existing) file + # subprocess_command += ["--save", "1 1"] + subprocess_command += [ + "--appendonly", + "yes", + "--appendfsync", + self.redis_fsync_frequency, + "--appenddirname", + self.aof_dir, + ] + logger.info("Redis persistence directory set to {}".format(self.aof_dir)) + elif ( + self.redis_saving_enabled + ): # just use the (possibly preexisting) default aof dir + subprocess_command += [ + "--appendonly", + "yes", + "--appendfsync", + self.redis_fsync_frequency, + ] + logger.info("Proceeding with using default Redis dump file.") + + logger.info( + "Starting Redis server with command: \n {}".format(subprocess_command) + ) + + return subprocess.Popen( + subprocess_command, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) def _closeStoreInterface(self): """Internal method to kill the subprocess running the store (plasma sever) """ - try: - self.p_StoreInterface.kill() - self.p_StoreInterface.wait() - logger.info("StoreInterface close successful: {}".format(self.store_loc)) - except Exception as e: - logger.exception("Cannot close store {}".format(e)) + if hasattr(self, "p_StoreInterface"): + try: + self.p_StoreInterface.send_signal(signal.SIGINT) + self.p_StoreInterface.wait() + logger.info( + "StoreInterface close successful: {}".format( + self.store_loc + if self.config and self.config.use_plasma() + else self.store_port + ) + ) + + except Exception as e: + logger.exception("Cannot close store {}".format(e)) def createActor(self, name, actor): """Function to instantiate actor, add signal and comm Links, @@ -642,7 +818,10 @@ def createActor(self, name, actor): # Instantiate selected class mod = import_module(actor.packagename) clss = getattr(mod, actor.classname) - instance = clss(actor.name, self.store_loc, **actor.options) + if self.config.use_plasma(): + instance = clss(actor.name, store_loc=self.store_loc, **actor.options) + else: + instance = clss(actor.name, store_port_num=self.store_port, **actor.options) if "method" in actor.options.keys(): # check for spawn @@ -720,7 +899,6 @@ def startWatcher(self): from improv.watcher import Watcher self.watcher = Watcher("watcher", self.createStoreInterface("watcher")) - # store = self.createStoreInterface("watcher") if not self.use_hdd else None q_sig = Link("watcher_sig", self.name, "watcher") self.watcher.setLinks(q_sig) self.sig_queues.update({q_sig.name: q_sig}) diff --git a/improv/replayer.py b/improv/replayer.py deleted file mode 100755 index 31044f8e7..000000000 --- a/improv/replayer.py +++ /dev/null @@ -1,160 +0,0 @@ -import time -from typing import Callable, List - -from pyarrow.plasma import ObjectID - -from nexus.actor import Actor, RunManager -from nexus.store import LMDBStoreInterface, LMDBData - - -class Replayer(Actor): - def __init__(self, *args, lmdb_path, replay: str, resave=False, **kwargs): - """ - Class that outputs objects to queues based on a saved previous run. - - Args: - lmdb_path: path to LMDB folder - replay: named of Actor to replay. - resave: (if using LMDB in this instance) - save outputs from this actor as usual (default: False) - - """ - super().__init__(*args, **kwargs) - self.resave = resave - - self.lmdb = LMDBStoreInterface(path=lmdb_path, load=True) - self.lmdb_values: list = self.get_lmdb_values(replay) - assert len(self.lmdb_values) > 0 - - self.gui_messages: dict = self.get_lmdb_values( - "GUI", func=lambda x: {lmdbdata.obj[0]: lmdbdata.time for lmdbdata in x} - ) - - self.t_saved_start_run = self.gui_messages["run"] # TODO Add load GUI actions - self.t_start_run = None - - def get_lmdb_values(self, replay: str, func: Callable = None) -> List[LMDBData]: - """ - Load saved queue objects from LMDB - - Args: - replay: named of Actor - func: (optional) Function to apply to objects before returning - - Returns: - lmdb_values - - """ - # Get all out queue names - replay = f"q__{replay}" - keys = [ - key.decode() - for key in self.lmdb.get_keys() - if key.startswith(replay.encode()) - ] - - # Get relevant keys, convert to str, and sort. Then convert back to bytes. - keys = [key.encode() for key in keys] - lmdb_values = sorted( - self.lmdb.get(keys, include_metadata=True), - key=lambda lmdb_value: lmdb_value.time, - ) - - if func is not None: - return func(lmdb_values) - return lmdb_values - - def setup(self): - if self.client.use_hdd and not self.resave: - self.client.use_hdd = False - - self.move_to_plasma(self.lmdb_values) - self.put_setup(self.lmdb_values) - - def move_to_plasma(self, lmdb_values): - """Put objects into current plasma store and update object ID in saved queue. - - Args: - lmdb_values: - """ - - # TODO Make async to enable queue-based fetch system - - # to avoid loading everything at once. - for lmdbdata in lmdb_values: - try: - if len(lmdbdata.obj) == 1 and isinstance( - lmdbdata.obj[0], dict - ): # Raw frames - data = lmdbdata.obj[0] - for i, obj_id in data.items(): - if isinstance(obj_id, ObjectID): - actual_obj = self.lmdb.get(obj_id, include_metadata=True) - lmdbdata.obj = [ - {i: self.client.put(actual_obj.obj, actual_obj.name)} - ] - - for i, obj in enumerate(lmdbdata.obj): # List - if isinstance(obj, ObjectID): - actual_obj = self.lmdb.get(obj, include_metadata=True) - lmdbdata.obj[i] = self.client.put( - actual_obj.obj, actual_obj.name - ) - - else: # Not object ID, do nothing. - pass - - except (TypeError, AttributeError): # Something else. - pass - - def put_setup(self, lmdb_values): - """Put all objects created before Run into queue immediately. - - Args: - lmdb_values: - """ - for lmdb_value in lmdb_values: - if lmdb_value.time < self.t_saved_start_run: - getattr(self, lmdb_value.queue).put(lmdb_value.obj) - - def run(self): - self.t_start_run = time.time() - with RunManager( - self.name, self.runner, self.setup, self.q_sig, self.q_comm - ) as rm: - print(rm) - - def runner(self): - """ - Get list of objects and output them - to their respective queues based on time delay. - """ - for lmdb_value in self.lmdb_values: - if lmdb_value.time >= self.t_saved_start_run: - t_sleep = ( - lmdb_value.time + self.t_start_run - self.t_saved_start_run - ) - time.time() - if t_sleep > 0: - time.sleep(t_sleep) - getattr(self, lmdb_value.queue).put(lmdb_value.obj) - - # policy = asyncio.get_event_loop_policy() - # policy.set_event_loop(policy.new_event_loop()) - # self.loop = asyncio.get_event_loop() - # - # self.aqueue = asyncio.Queue() - # self.loop.run_until_complete(self.arun()) - # - # async def arun(self): - # - # - # funcs_to_run = [self.send_q, self.fetch_lmdb] - # async with AsyncRunManager(self.name, funcs_to_run, self.setup, - # self.q_sig, self.q_comm) as rm: - # print(rm) - # - # async def send_q(self): - # - # for t in self.times: - # now = time.time() - # await asyncio.sleep(t - now) - # self.q_out.put(list(dict())) diff --git a/improv/store.py b/improv/store.py index d6c4b62cd..98f54a63a 100644 --- a/improv/store.py +++ b/improv/store.py @@ -1,22 +1,23 @@ -import lmdb -import time +import os +import uuid + import pickle -import signal import logging import traceback import numpy as np import pyarrow.plasma as plasma -from queue import Queue -from pathlib import Path -from random import random -from threading import Thread -from typing import List, Union -from dataclasses import dataclass, make_dataclass +from redis import Redis +from redis.retry import Retry +from redis.backoff import ConstantBackoff +from redis.exceptions import BusyLoadingError, ConnectionError, TimeoutError + from scipy.sparse import csc_matrix from pyarrow.lib import ArrowIOError -from pyarrow._plasma import PlasmaObjectExists, ObjectNotAvailable, ObjectID +from pyarrow._plasma import PlasmaObjectExists, ObjectNotAvailable + +REDIS_GLOBAL_TOPIC = "global_topic" logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) @@ -41,6 +42,143 @@ def subscribe(self): raise NotImplementedError +class RedisStoreInterface(StoreInterface): + def __init__(self, name="default", server_port_num=6379, hostname="localhost"): + self.name = name + self.server_port_num = server_port_num + self.hostname = hostname + self.client = self.connect_to_server() + + def connect_to_server(self): + # TODO this should scan for available ports, but only if configured to do so. + # This happens when the config doesn't have Redis settings, + # so we need to communicate this somehow to the StoreInterface here. + """Connect to the store at store_loc, max 20 retries to connect + Raises exception if can't connect + Returns the Redis client if successful + + Args: + server_port_num: the port number where the Redis server + is running on localhost. + """ + try: + retry = Retry(ConstantBackoff(0.25), 5) + self.client = Redis( + host=self.hostname, + port=self.server_port_num, + retry=retry, + retry_on_timeout=True, + retry_on_error=[ + BusyLoadingError, + ConnectionError, + TimeoutError, + ConnectionRefusedError, + ], + ) + self.client.ping() + logger.info( + "Successfully connected to redis datastore on port {} ".format( + self.server_port_num + ) + ) + except Exception: + logger.exception( + "Cannot connect to redis datastore on port {}".format( + self.server_port_num + ) + ) + raise CannotConnectToStoreInterfaceError(self.server_port_num) + + return self.client + + def put(self, object): + """ + Put a single object referenced by its string name + into the store. If the store already has a value stored at this key, + the value will not be overwritten. + + Unknown error + + Args: + object: the object to store in Redis + object_key (str): the key under which the object should be stored + + Returns: + object: the object that was a + """ + object_key = str(os.getpid()) + str(uuid.uuid4()) + try: + # buffers would theoretically go here if we need to force out-of-band + # serialization for single objects + # TODO this will actually just silently fail if we use an existing + # TODO key; not sure it's worth the network overhead to check every + # TODO key twice every time. we still need a better solution for + # TODO this, but it will work now singlethreaded most of the time. + + self.client.set(object_key, pickle.dumps(object, protocol=5), nx=True) + except Exception: + logger.error("Could not store object {}".format(object_key)) + logger.error(traceback.format_exc()) + + return object_key + + def get(self, object_key): + """ + Get object by specified key + + Args: + object_name: the key of the object + + Returns: + Stored object + + Raises: + ObjectNotFoundError: If the key is not found + """ + object_value = self.client.get(object_key) + if object_value: + # buffers would also go here to force out-of-band deserialization + return pickle.loads(object_value) + + logger.warning("Object {} cannot be found.".format(object_key)) + raise ObjectNotFoundError + + def subscribe(self, topic=REDIS_GLOBAL_TOPIC): + p = self.client.pubsub() + p.subscribe(topic) + + def get_list(self, ids): + """Get multiple objects from the store + + Args: + ids (list): of type str + + Returns: + list of the objects + """ + return self.client.mget(ids) + + def get_all(self): + """Get a listing of all objects in the store. + Note that this may be very performance-intensive in large databases. + + Returns: + list of all the objects in the store + """ + all_keys = self.client.keys() # defaults to "*" pattern, so will fetch all + return self.client.mget(all_keys) + + def reset(self): + """Reset client connection""" + self.client = self.connect_to_server() + logger.debug( + "Reset local connection to store on port: {0}".format(self.server_port_num) + ) + + def notify(self): + pass # I don't see any call sites for this, so leaving it blank at the moment + + class PlasmaStoreInterface(StoreInterface): """Basic interface for our specific data store implemented with apache arrow plasma Objects are stored with object_ids @@ -60,49 +198,6 @@ def __init__(self, name="default", store_loc="/tmp/store"): self.store_loc = store_loc self.client = self.connect_store(store_loc) self.stored = {} - self.use_hdd = False - - def _setup_LMDB( - self, - use_lmdb=False, - lmdb_path="../outputs/", - lmdb_name=None, - hdd_maxstore=1e12, - flush_immediately=False, - commit_freq=1, - ): - """ - Constructor for the Store - - Args: - name (string): - store_loc (stirng): Apache Arrow Plasma client location - - use_lmdb (bool): Also write data to disk using the LMDB - - hdd_maxstore: - Maximum size database may grow to; used to size the memory mapping. - If the database grows larger than map_size, - a MapFullError will be raised. - On 64-bit there is no penalty for making this huge. - Must be <2GB on 32-bit. - - hdd_path: Path to LMDB folder. - flush_immediately (bool): Save objects to disk immediately - commit_freq (int): If not flush_immediately, - flush data to disk every {commit_freq} seconds. - """ - self.use_hdd = use_lmdb - self.flush_immediately = flush_immediately - - if use_lmdb: - self.lmdb_store = LMDBStoreInterface( - path=lmdb_path, - name=lmdb_name, - max_size=hdd_maxstore, - flush_immediately=flush_immediately, - commit_freq=commit_freq, - ) def connect_store(self, store_loc): """Connect to the store at store_loc, max 20 retries to connect @@ -149,8 +244,6 @@ class 'plasma.ObjectID': Plasma object ID else: object_id = self.client.put(object) - if self.use_hdd: - self.lmdb_store.put(object, object_name, obj_id=object_id) except PlasmaObjectExists: logger.error("Object already exists. Meant to call replace?") except ArrowIOError: @@ -181,13 +274,12 @@ def get(self, object_name): # else: return self.getID(object_name) - def getID(self, obj_id, hdd_only=False): + def getID(self, obj_id): """ Get object by object ID Args: obj_id (class 'plasma.ObjectID'): the id of the object - hdd_only (bool): Returns: Stored object @@ -195,17 +287,9 @@ def getID(self, obj_id, hdd_only=False): Raises: ObjectNotFoundError: If the id is not found """ - # Check in RAM - if not hdd_only: - res = self.client.get(obj_id, 0) # Timeout = 0 ms - if res is not plasma.ObjectNotAvailable: - return res if not isinstance(res, bytes) else pickle.loads(res) - - # Check in disk - if self.use_hdd: - res = self.lmdb_store.get(obj_id) - if res is not None: - return res + res = self.client.get(obj_id, 0) # Timeout = 0 ms + if res is not plasma.ObjectNotAvailable: + return res if not isinstance(res, bytes) else pickle.loads(res) logger.warning("Object {} cannot be found.".format(obj_id)) raise ObjectNotFoundError @@ -312,239 +396,7 @@ def _get(self, object_name): return res -class LMDBStoreInterface(StoreInterface): - def __init__( - self, - path="../outputs/", - name=None, - load=False, - max_size=1e12, - flush_immediately=False, - commit_freq=1, - ): - """ - Constructor for LMDB store - - Args: - path (string): Path to folder containing LMDB folder. - name (string): Name of LMDB. Required if not {load]. - max_size (float): - Maximum size database may grow to; used to size the memory mapping. - If the database grows larger than map_size, a MapFullError will be raised. - On 64-bit there is no penalty for making this huge. Must be <2GB on 32-bit. - load (bool): For Replayer use. Informs the class that we're loading - from a previous LMDB, not create a new one. - flush_immediately (bool): Save objects to disk immediately - commit_freq (int): If not flush_immediately, - flush data to disk every {commit_freq} seconds. - """ - - # Check if LMDB folder exists. - # LMDB path? - path = Path(path) - if load: - if name is not None: - path = path / name - if not (path / "data.mdb").exists(): - raise FileNotFoundError("Invalid LMDB directory.") - else: - assert name is not None - if not path.exists(): - path.mkdir(parents=True) - path = path / name - - self.flush_immediately = flush_immediately - self.lmdb_env = lmdb.open( - path.as_posix(), map_size=max_size, sync=flush_immediately - ) - self.lmdb_commit_freq = commit_freq - - self.put_queue = Queue() - self.put_queue_container = make_dataclass( - "LMDBPutContainer", [("name", str), ("obj", bytes)] - ) - # Initialize only after interpreter has forked at the start of each actor. - self.commit_thread: Thread = None - signal.signal(signal.SIGINT, self.flush) - - def get( - self, - key: Union[plasma.ObjectID, bytes, List[plasma.ObjectID], List[bytes]], - include_metadata=False, - ): - """ - Get object using key (could be any byte string or plasma.ObjectID) - - Args: - key: - include_metadata (bool): returns whole LMDBData if true else LMDBData.obj - (just the stored object). - Returns: - object or LMDBData - """ - while True: - try: - if isinstance(key, str) or isinstance(key, ObjectID): - return self._get_one( - LMDBStoreInterface._convert_obj_id_to_bytes(key), - include_metadata, - ) - return self._get_batch( - list(map(LMDBStoreInterface._convert_obj_id_to_bytes, key)), - include_metadata, - ) - except ( - lmdb.BadRslotError - ): # Happens when multiple transactions access LMDB at the same time. - pass - - def _get_one(self, key, include_metadata): - with self.lmdb_env.begin() as txn: - r = txn.get(key) - - if r is None: - return None - return pickle.loads(r) if include_metadata else pickle.loads(r).obj - - def _get_batch(self, keys, include_metadata): - with self.lmdb_env.begin() as txn: - objs = [txn.get(key) for key in keys] - - if include_metadata: - return [pickle.loads(obj) for obj in objs if obj is not None] - else: - return [pickle.loads(obj).obj for obj in objs if obj is not None] - - def get_keys(self): - """Get all keys in LMDB""" - with self.lmdb_env.begin() as txn: - with txn.cursor() as cur: - cur.first() - return [key for key in cur.iternext(values=False)] - - def put(self, obj, obj_name, obj_id=None, flush_this_immediately=False): - """ - Put object ID / object pair into LMDB. - - Args: - obj: Object to be saved - obj_name (str): the name of the object - obj_id ('plasma.ObjectID'): Object_id from Plasma client - flush_this_immediately (bool): Override self.flush_immediately. - For storage of critical objects. - - Returns: - None - """ - # TODO: Duplication check - if self.commit_thread is None: - self.commit_thread = Thread(target=self.commit_daemon, daemon=True) - self.commit_thread.start() - - if obj_name.startswith("q_") or obj_name.startswith("config"): # Queue - name = obj_name.encode() - is_queue = True - else: - name = obj_id.binary() if obj_id is not None else obj_name.encode() - is_queue = False - - self.put_queue.put( - self.put_queue_container( - name=name, - obj=pickle.dumps( - LMDBData(obj, time=time.time(), name=obj_name, is_queue=is_queue) - ), - ) - ) - - # Write - if self.flush_immediately or flush_this_immediately: - self.commit() - self.lmdb_env.sync() - - def flush(self, sig=None, frame=None): - """Must run before exiting. Flushes buffer to disk.""" - self.commit() - self.lmdb_env.sync() - self.lmdb_env.close() - exit(0) - - def commit_daemon(self): - time.sleep(2 * random()) # Reduce multiple commits at the same time. - while True: - time.sleep(self.lmdb_commit_freq) - self.commit() - - def commit(self): - """Commit objects in {self.put_cache} into LMDB.""" - if not self.put_queue.empty(): - print(self.put_queue.qsize()) - with self.lmdb_env.begin(write=True) as txn: - while not self.put_queue.empty(): - container = self.put_queue.get_nowait() - txn.put(container.name, container.obj, overwrite=True) - - def delete(self, obj_id): - """ - Delete object from LMDB. - - Args: - obj_id (class 'plasma.ObjectID'): the object_id to be deleted - - Returns: - None - - Raises: - ObjectNotFoundError: If the id is not found - """ - with self.lmdb_env.begin(write=True) as txn: - out = txn.pop(LMDBStoreInterface._convert_obj_id_to_bytes(obj_id)) - if out is None: - raise ObjectNotFoundError - - @staticmethod - def _convert_obj_id_to_bytes(obj_id): - try: - return obj_id.binary() - except AttributeError: - return obj_id - - def replace(self): - pass # TODO - - def subscribe(self): - pass # TODO - - -# Aliasing -StoreInterface = PlasmaStoreInterface - - -@dataclass -class LMDBData: - """ - Dataclass to store objects and their metadata into LMDB. - """ - - obj: object - time: float - name: str = None - is_queue: bool = False - - @property - def queue(self): - """ - Returns: - Queue name if object is a queue else None - """ - # Expected: 'q__Acquirer.q_out__124' -> {'q_out'} - if self.is_queue: - try: - return self.name.split("__")[1].split(".")[1] - except IndexError: - return "q_comm" - logger.error("Attempt to get queue name from objects not from queue.") - return None +StoreInterface = RedisStoreInterface class ObjectNotFoundError(Exception): diff --git a/improv/utils/bad_config.yaml b/improv/utils/bad_config.yaml deleted file mode 100644 index d30cfd56c..000000000 --- a/improv/utils/bad_config.yaml +++ /dev/null @@ -1,40 +0,0 @@ -modules: - GUI: - package: visual.visual - class: DisplayVisual - visual: Visual - - Acquirer: - package: acquire.acquire - class: LMDBAcquirer - lmdb_path: /Users/chaichontat/Desktop/Lab/New/rasp/src/output/lmdb - # filename: /Users/chaichontat/Desktop/Lab/New/rasp/data/Tolias_mesoscope_2.hdf5 - framerate: 15 - - Processor: - package: process.process - class: CaimanProcessor - - Visual: - package: visual.visual - class: CaimanVisual - - Analysis: - package: analysis.analysis - class: MeanAnalysis - - InputStim: - package: acquire.acquire - class: BehaviorAcquirer - - Watcherr: - package: watch.watch - class: Watcher - - -connections: - Acquirer.q_out: [Processor.q_in, Visual.raw_frame_queue, Watcherr.raw_frame_queue] - Processor.q_out: [Analysis.q_in] - Analysis.q_out: [Visual.q_in, Acquirer.q_in] - InputStim.q_out: [Analysis.input_stim_queue] - diff --git a/improv/utils/checks.py b/improv/utils/checks.py index 230107cd3..a72f46572 100644 --- a/improv/utils/checks.py +++ b/improv/utils/checks.py @@ -5,13 +5,6 @@ $ python checks.py [file_name].yaml - $ python checks.py good_config.yaml - No loops. - - $ python checks.py bad_config.yaml - Loop(s) found. - Processor to Analysis to Acquirer - """ import sys diff --git a/improv/utils/good_config.yaml b/improv/utils/good_config.yaml deleted file mode 100644 index d91764c81..000000000 --- a/improv/utils/good_config.yaml +++ /dev/null @@ -1,40 +0,0 @@ -modules: - GUI: - package: visual.visual - class: DisplayVisual - visual: Visual - - Acquirer: - package: acquire.acquire - class: LMDBAcquirer - lmdb_path: /Users/chaichontat/Desktop/Lab/New/rasp/src/output/lmdb - # filename: /Users/chaichontat/Desktop/Lab/New/rasp/data/Tolias_mesoscope_2.hdf5 - framerate: 15 - - Processor: - package: process.process - class: CaimanProcessor - - Visual: - package: visual.visual - class: CaimanVisual - - Analysis: - package: analysis.analysis - class: MeanAnalysis - - InputStim: - package: acquire.acquire - class: BehaviorAcquirer - - Watcherr: - package: watch.watch - class: Watcher - - -connections: - Acquirer.q_out: [Processor.q_in, Visual.raw_frame_queue, Watcherr.raw_frame_queue] - Processor.q_out: [Analysis.q_in] - Analysis.q_out: [Visual.q_in] - InputStim.q_out: [Analysis.input_stim_queue] - diff --git a/improv/utils/reader.py b/improv/utils/reader.py deleted file mode 100644 index 036d8f3db..000000000 --- a/improv/utils/reader.py +++ /dev/null @@ -1,104 +0,0 @@ -import os -import pickle -from contextlib import contextmanager - -# from typing import Dict, Set -import lmdb -from .utils import get_num_length_from_key - - -class LMDBReader: - def __init__(self, path): - """Constructor for the LMDB reader - path: Path to LMDB folder - """ - if not os.path.exists(path): - raise FileNotFoundError - self.path = path - - def get_all_data(self): - """Load all data from LMDB into a dictionary - Make sure that the LMDB is small enough to fit in RAM - """ - with LMDBReader._lmdb_cur(self.path) as cur: - return { - LMDBReader._decode_key(key): pickle.loads(value) - for key, value in cur.iternext() - } - - def get_data_types(self): - """Return all data types defined as {object_name}, but without number.""" - num_idx = get_num_length_from_key() - - with LMDBReader._lmdb_cur(self.path) as cur: - return { - key[: -12 - num_idx.send(key)] for key in cur.iternext(values=False) - } - - def get_data_by_number(self, t): - """Return data at a specific frame number t""" - num_idx = get_num_length_from_key() - - def check_if_key_equals_t(key): - try: - return True if int(key[-12 - num_idx.send(key) : -12]) == t else False - except ValueError: - return False - - with LMDBReader._lmdb_cur(self.path) as cur: - keys = ( - key for key in cur.iternext(values=False) if check_if_key_equals_t(key) - ) - return { - LMDBReader._decode_key(key): pickle.loads(cur.get(key)) for key in keys - } - - def get_data_by_type(self, t): - """Return data with key that starts with t""" - with LMDBReader._lmdb_cur(self.path) as cur: - keys = ( - key for key in cur.iternext(values=False) if key.startswith(t.encode()) - ) - return { - LMDBReader._decode_key(key): pickle.loads(cur.get(key)) for key in keys - } - - def get_params(self): - """Return parameters in a dictionary""" - with LMDBReader._lmdb_cur(self.path) as cur: - keys = [ - key - for key in cur.iternext(values=False) - if key.startswith(b"params_dict") - ] - return pickle.loads(cur.get(keys[-1])) - - @staticmethod - def _decode_key(key): - """Helper method to convert key from byte to str - - Example: - >>> LMDBReader._decode_key(b'Call0\x80\x03GA\xd7Ky\x06\x9c\xddi.') - 'Call0_1563288602.4510138' - - key: Encoded key. The last 12 bytes are pickled time.time(). - The remaining are encoded object name. - """ - - return f"{key[:-12].decode()}_{pickle.loads(key[-12:])}" - - @staticmethod - @contextmanager - def _lmdb_cur(path): - """Helper context manager to open and ensure proper closure of LMDB""" - - env = lmdb.open(path) - txn = env.begin() - cur = txn.cursor() - try: - yield cur - - finally: - cur.__exit__() - txn.commit() - env.close() diff --git a/improv/utils/utils.py b/improv/utils/utils.py deleted file mode 100644 index c7cc00723..000000000 --- a/improv/utils/utils.py +++ /dev/null @@ -1,51 +0,0 @@ -# from functools import wraps - -# def coroutine(func): #FIXME who uses this and why? -# """ Decorator that primes 'func' by calling first {yield}. """ - -# @wraps(func) -# def primer(*args, **kwargs): -# gen = func(*args, **kwargs) -# next(gen) -# return gen -# return primer - - -# @coroutine -def get_num_length_from_key(): - """ - Coroutine that gets the length of digits in LMDB key. - Assumes that object name does not have any digits. - - For example: - FileAcquirer puts objects with names 'acq_raw{i}' where i is the frame number. - {i}, however, is not padded with zero, so the length changes with number. - The B-tree sorting in LMDB results in messed up number sorting. - Example: - >>> num_idx = get_num_length_from_key() - >>> num_idx.send(b'acq_raw1\x80\x03GA\xd7L\x1b\x8f\xb0\x1b\xb0.') - 1 - - """ - max_num_len = 1 # Keep track of largest digit for performance. - - def worker(): - nonlocal max_num_len - name_num = key[:-12].decode() - - if not name_num[-max_num_len:].isdigit(): - i = max_num_len - while not name_num[-i:].isdigit(): - if i < 1: - return 0 - i -= 1 - return i - - while name_num[-(max_num_len + 1) :].isdigit(): - max_num_len += 1 - return max_num_len - - num = "Primed!" - while True: - key = yield num - num = worker() diff --git a/pyproject.toml b/pyproject.toml index ad2efae34..bf5a42b78 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,26 +6,39 @@ build-backend = "setuptools.build_meta" name = "improv" description = "Platform for adaptive neuroscience experiments" authors = [{name = "Anne Draelos", email = "adraelos@umich.edu"}, - {name = "John Pearson", email = "john.pearson@duke.edu"}] + {name = "John Pearson", email = "john.pearson@duke.edu"}] license = {file = "LICENSE"} readme = "README.md" requires-python = ">=3.6" keywords = ["neuroscience", "adaptive", "closed loop"] dependencies = [ - "numpy", + "numpy<=1.26", "scipy", "matplotlib", "pyarrow==9.0.0", "PyQt5", - "lmdb", "pyyaml", "textual==0.15.0", "pyzmq", "psutil", "h5py", + "redis" ] -classifiers = ['Development Status :: 1 - Planning'] +classifiers = ['Development Status :: 3 - Alpha', + 'Intended Audience :: Science/Research', + 'License :: OSI Approved :: MIT License', + 'Operating System :: MacOS :: MacOS X', + 'Operating System :: Microsoft :: Windows', + 'Operating System :: POSIX', + 'Programming Language :: Python :: 3.7', + 'Programming Language :: Python :: 3.8', + 'Programming Language :: Python :: 3.9', + 'Programming Language :: Python :: 3.10', + 'Programming Language :: Python :: 3.11', + 'Programming Language :: Python :: 3.12', + 'Programming Language :: Python :: Implementation :: CPython' + ] dynamic = ["version"] [project.optional-dependencies] @@ -99,4 +112,4 @@ distance-dirty = "{next_version}.dev{distance}+{vcs}{rev}.d{build_date:%Y%m%d}" # Example formatted version: 1.2.4.dev42+ge174a1f.d20230922 [tool.versioningit.write] -file = "improv/_version.py" \ No newline at end of file +file = "improv/_version.py" diff --git a/test/configs/good_config_plasma.yaml b/test/configs/good_config_plasma.yaml new file mode 100644 index 000000000..8323017b1 --- /dev/null +++ b/test/configs/good_config_plasma.yaml @@ -0,0 +1,15 @@ +actors: + Acquirer: + package: demos.sample_actors.acquire + class: FileAcquirer + filename: data/Tolias_mesoscope_2.hdf5 + framerate: 30 + + Analysis: + package: demos.sample_actors.simple_analysis + class: SimpleAnalysis + +connections: + Acquirer.q_out: [Analysis.q_in] + +plasma_config: \ No newline at end of file diff --git a/test/configs/minimal_with_custom_aof_dirname.yaml b/test/configs/minimal_with_custom_aof_dirname.yaml new file mode 100644 index 000000000..1bac58635 --- /dev/null +++ b/test/configs/minimal_with_custom_aof_dirname.yaml @@ -0,0 +1,14 @@ +actors: + Generator: + package: actors.sample_generator + class: Generator + + Processor: + package: actors.sample_processor + class: Processor + +connections: + Generator.q_out: [Processor.q_in] + +redis_config: + aof_dirname: custom_aof_dirname \ No newline at end of file diff --git a/test/configs/minimal_with_ephemeral_aof_dirname.yaml b/test/configs/minimal_with_ephemeral_aof_dirname.yaml new file mode 100644 index 000000000..9933896ee --- /dev/null +++ b/test/configs/minimal_with_ephemeral_aof_dirname.yaml @@ -0,0 +1,14 @@ +actors: + Generator: + package: actors.sample_generator + class: Generator + + Processor: + package: actors.sample_processor + class: Processor + +connections: + Generator.q_out: [Processor.q_in] + +redis_config: + generate_ephemeral_aof_dirname: True \ No newline at end of file diff --git a/test/configs/minimal_with_every_second_saving.yaml b/test/configs/minimal_with_every_second_saving.yaml new file mode 100644 index 000000000..74d4b314b --- /dev/null +++ b/test/configs/minimal_with_every_second_saving.yaml @@ -0,0 +1,15 @@ +actors: + Generator: + package: actors.sample_generator + class: Generator + + Processor: + package: actors.sample_processor + class: Processor + +connections: + Generator.q_out: [Processor.q_in] + +redis_config: + enable_saving: True + fsync_frequency: every_second \ No newline at end of file diff --git a/test/configs/minimal_with_every_write_saving.yaml b/test/configs/minimal_with_every_write_saving.yaml new file mode 100644 index 000000000..add0c74fd --- /dev/null +++ b/test/configs/minimal_with_every_write_saving.yaml @@ -0,0 +1,15 @@ +actors: + Generator: + package: actors.sample_generator + class: Generator + + Processor: + package: actors.sample_processor + class: Processor + +connections: + Generator.q_out: [Processor.q_in] + +redis_config: + enable_saving: True + fsync_frequency: every_write \ No newline at end of file diff --git a/test/configs/minimal_with_fixed_default_redis_port.yaml b/test/configs/minimal_with_fixed_default_redis_port.yaml new file mode 100644 index 000000000..c8e0b24ee --- /dev/null +++ b/test/configs/minimal_with_fixed_default_redis_port.yaml @@ -0,0 +1,14 @@ +actors: + Generator: + package: actors.sample_generator + class: Generator + + Processor: + package: actors.sample_processor + class: Processor + +connections: + Generator.q_out: [Processor.q_in] + +redis_config: + port: 6379 \ No newline at end of file diff --git a/test/configs/minimal_with_fixed_redis_port.yaml b/test/configs/minimal_with_fixed_redis_port.yaml new file mode 100644 index 000000000..f6f982536 --- /dev/null +++ b/test/configs/minimal_with_fixed_redis_port.yaml @@ -0,0 +1,14 @@ +actors: + Generator: + package: actors.sample_generator + class: Generator + + Processor: + package: actors.sample_processor + class: Processor + +connections: + Generator.q_out: [Processor.q_in] + +redis_config: + port: 6378 \ No newline at end of file diff --git a/test/configs/minimal_with_no_schedule_saving.yaml b/test/configs/minimal_with_no_schedule_saving.yaml new file mode 100644 index 000000000..2e5f62f34 --- /dev/null +++ b/test/configs/minimal_with_no_schedule_saving.yaml @@ -0,0 +1,15 @@ +actors: + Generator: + package: actors.sample_generator + class: Generator + + Processor: + package: actors.sample_processor + class: Processor + +connections: + Generator.q_out: [Processor.q_in] + +redis_config: + enable_saving: True + fsync_frequency: no_schedule \ No newline at end of file diff --git a/test/configs/minimal_with_redis_saving.yaml b/test/configs/minimal_with_redis_saving.yaml new file mode 100644 index 000000000..15e95b6c9 --- /dev/null +++ b/test/configs/minimal_with_redis_saving.yaml @@ -0,0 +1,14 @@ +actors: + Generator: + package: actors.sample_generator + class: Generator + + Processor: + package: actors.sample_processor + class: Processor + +connections: + Generator.q_out: [Processor.q_in] + +redis_config: + enable_saving: True \ No newline at end of file diff --git a/test/configs/minimal_with_settings.yaml b/test/configs/minimal_with_settings.yaml index 37d43c55f..a5aec2d8a 100644 --- a/test/configs/minimal_with_settings.yaml +++ b/test/configs/minimal_with_settings.yaml @@ -4,7 +4,6 @@ settings: control_port: 6000 output_port: 6001 logging_port: 6002 - use_hdd: false use_watcher: false actors: diff --git a/test/configs/single_actor_plasma.yaml b/test/configs/single_actor_plasma.yaml new file mode 100644 index 000000000..812480b12 --- /dev/null +++ b/test/configs/single_actor_plasma.yaml @@ -0,0 +1,10 @@ +actors: + Acquirer: + package: demos.sample_actors.acquire + class: FileAcquirer + filename: data/Tolias_mesoscope_2.hdf5 + framerate: 15 + +connections: +# settings: +# use_watcher: [Acquirer, Processor, Visual, Analysis] diff --git a/test/conftest.py b/test/conftest.py index 5bbdb5526..f9b80fa5b 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -1,10 +1,57 @@ import os +import signal import uuid import pytest +import subprocess store_loc = str(os.path.join("/tmp/", str(uuid.uuid4()))) +redis_port_num = 6379 +WAIT_TIMEOUT = 120 -@pytest.fixture() +@pytest.fixture def set_store_loc(): return store_loc + + +@pytest.fixture +def server_port_num(): + return redis_port_num + + +@pytest.fixture +# TODO: put in conftest.py +def setup_store(server_port_num): + """Start the server""" + p = subprocess.Popen( + [ + "redis-server", + "--save", + '""', + "--port", + str(server_port_num), + "--maxmemory", + str(10000000), + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + yield p + + # kill the subprocess when the caller is done with it + p.send_signal(signal.SIGINT) + p.wait(WAIT_TIMEOUT) + + +@pytest.fixture +def setup_plasma_store(set_store_loc, scope="module"): + """Fixture to set up the store subprocess with 10 mb.""" + p = subprocess.Popen( + ["plasma_store", "-s", set_store_loc, "-m", str(10000000)], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + yield p + p.send_signal(signal.SIGINT) + p.wait(WAIT_TIMEOUT) diff --git a/test/test_actor.py b/test/test_actor.py index 0745aa698..448fd34e1 100644 --- a/test/test_actor.py +++ b/test/test_actor.py @@ -1,11 +1,9 @@ import os import psutil import pytest -import subprocess from improv.link import Link # , AsyncQueue from improv.actor import AbstractActor as Actor -from improv.store import StoreInterface - +from improv.store import StoreInterface, PlasmaStoreInterface # set global_variables @@ -13,20 +11,7 @@ pytest.example_links = {} -@pytest.fixture() -def setup_store(set_store_loc, scope="module"): - """Fixture to set up the store subprocess with 10 mb.""" - p = subprocess.Popen( - ["plasma_store", "-s", set_store_loc, "-m", str(10000000)], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) - yield p - p.kill() - p.wait() - - -@pytest.fixture() +@pytest.fixture def init_actor(set_store_loc): """Fixture to initialize and teardown an instance of actor.""" @@ -35,7 +20,7 @@ def init_actor(set_store_loc): act = None -@pytest.fixture() +@pytest.fixture def example_string_links(): """Fixture to provide a commonly used test input.""" @@ -43,10 +28,27 @@ def example_string_links(): return pytest.example_string_links -@pytest.fixture() -def example_links(setup_store, set_store_loc): +@pytest.fixture +def example_links(setup_store, server_port_num): """Fixture to provide link objects as test input and setup store.""" - StoreInterface(store_loc=set_store_loc) + StoreInterface(server_port_num=server_port_num) + + acts = [ + Actor("act" + str(i), server_port_num) for i in range(1, 5) + ] # range must be even + + links = [ + Link("L" + str(i + 1), acts[i], acts[i + 1]) for i in range(len(acts) // 2) + ] + link_dict = {links[i].name: links[i] for i, l in enumerate(links)} + pytest.example_links = link_dict + return pytest.example_links + + +@pytest.fixture +def example_links_plasma(setup_store, set_store_loc): + """Fixture to provide link objects as test input and setup store.""" + PlasmaStoreInterface(store_loc=set_store_loc) acts = [ Actor("act" + str(i), set_store_loc) for i in range(1, 5) @@ -95,11 +97,20 @@ def test_repr(example_string_links, set_store_loc): assert act.__repr__() == "Test: dict_keys(['1', '2', '3'])" -def test_setStoreInterface(setup_store, set_store_loc): +def test_setStoreInterface(setup_store, server_port_num): + """Tests if the store is started and linked with the actor.""" + + act = Actor("Acquirer", server_port_num) + store = StoreInterface(server_port_num=server_port_num) + act.setStoreInterface(store.client) + assert act.client is store.client + + +def test_plasma_setStoreInterface(setup_plasma_store, set_store_loc): """Tests if the store is started and linked with the actor.""" act = Actor("Acquirer", set_store_loc) - store = StoreInterface(store_loc=set_store_loc) + store = PlasmaStoreInterface(store_loc=set_store_loc) act.setStoreInterface(store.client) assert act.client is store.client @@ -292,7 +303,30 @@ def test_changePriority(init_actor): assert psutil.Process(os.getpid()).nice() == 19 -def test_actor_connection(setup_store, set_store_loc): +def test_actor_connection(setup_store, server_port_num): + """Test if the links between actors are established correctly. + + This test instantiates two actors with different names, then instantiates + a Link object linking the two actors. A string is put to the input queue of + one actor. Then, in the other actor, it is removed from the queue, and + checked to verify it matches the original message. + """ + act1 = Actor("a1", server_port_num) + act2 = Actor("a2", server_port_num) + + StoreInterface(server_port_num=server_port_num) + link = Link("L12", act1, act2) + act1.setLinkIn(link) + act2.setLinkOut(link) + + msg = "message" + + act1.q_in.put(msg) + + assert act2.q_out.get() == msg + + +def test_plasma_actor_connection(setup_plasma_store, set_store_loc): """Test if the links between actors are established correctly. This test instantiates two actors with different names, then instantiates @@ -303,7 +337,7 @@ def test_actor_connection(setup_store, set_store_loc): act1 = Actor("a1", set_store_loc) act2 = Actor("a2", set_store_loc) - StoreInterface(store_loc=set_store_loc) + PlasmaStoreInterface(store_loc=set_store_loc) link = Link("L12", act1, act2) act1.setLinkIn(link) act2.setLinkOut(link) diff --git a/test/test_cli.py b/test/test_cli.py index 7e18363fe..c4b34bff3 100644 --- a/test/test_cli.py +++ b/test/test_cli.py @@ -13,7 +13,7 @@ SERVER_TIMEOUT = 16 -@pytest.fixture() +@pytest.fixture def setdir(): prev = os.getcwd() os.chdir(os.path.dirname(__file__)) @@ -21,7 +21,7 @@ def setdir(): os.chdir(prev) -@pytest.fixture() +@pytest.fixture async def server(setdir, ports): """ Sets up a server using minimal.yaml in the configs folder. @@ -61,7 +61,7 @@ async def server(setdir, ports): pass -@pytest.fixture() +@pytest.fixture async def cli_args(setdir, ports): logfile = "tmp.log" control_port, output_port, logging_port = ports diff --git a/test/test_config.py b/test/test_config.py index f9d112018..12cc79bf5 100644 --- a/test/test_config.py +++ b/test/test_config.py @@ -16,7 +16,7 @@ # set global variables -@pytest.fixture() +@pytest.fixture def set_configdir(): """Sets the current working directory to the configs file.""" prev = os.getcwd() diff --git a/test/test_demos.py b/test/test_demos.py index 61098c679..bc6e96c67 100644 --- a/test/test_demos.py +++ b/test/test_demos.py @@ -1,3 +1,5 @@ +import time + import pytest import os import asyncio @@ -16,7 +18,7 @@ SERVER_WARMUP = 10 -@pytest.fixture() +@pytest.fixture def setdir(): prev = os.getcwd() os.chdir(os.path.dirname(__file__)) @@ -25,7 +27,7 @@ def setdir(): os.chdir(prev) -@pytest.fixture() +@pytest.fixture def ip(): """Fixture to provide an IP test input.""" @@ -34,7 +36,11 @@ def ip(): @pytest.mark.parametrize( - ("dir", "configfile", "logfile"), [("minimal", "minimal.yaml", "testlog")] + ("dir", "configfile", "logfile"), + [ + ("minimal", "minimal.yaml", "testlog"), + ("minimal", "minimal_plasma.yaml", "testlog"), + ], ) async def test_simple_boot_and_quit(dir, configfile, logfile, setdir, ports): os.chdir(dir) @@ -58,6 +64,8 @@ async def test_simple_boot_and_quit(dir, configfile, logfile, setdir, ports): with open(logfile, mode="a+") as log: server = subprocess.Popen(server_opts, stdout=log, stderr=log) + time.sleep(5) + print(log.readlines()) await asyncio.sleep(SERVER_WARMUP) # initialize client @@ -79,7 +87,10 @@ async def test_simple_boot_and_quit(dir, configfile, logfile, setdir, ports): @pytest.mark.parametrize( ("dir", "configfile", "logfile", "datafile"), - [("minimal", "minimal.yaml", "testlog", "sample_generator_data.npy")], + [ + ("minimal", "minimal.yaml", "testlog", "sample_generator_data.npy"), + ("minimal", "minimal_spawn.yaml", "testlog", "sample_generator_data.npy"), + ], ) async def test_stop_output(dir, configfile, logfile, datafile, setdir, ports): os.chdir(dir) @@ -136,8 +147,8 @@ def test_zmq_ps(ip, unused_tcp_port): """Tests if we can set the zmq PUB/SUB socket and send message.""" port = unused_tcp_port LOGGER.info("beginning test") - act1 = ZmqActor("act1", "/tmp/store", type="PUB", ip=ip, port=port) - act2 = ZmqActor("act2", "/tmp/store", type="SUB", ip=ip, port=port) + act1 = ZmqActor("act1", type="PUB", ip=ip, port=port) + act2 = ZmqActor("act2", type="SUB", ip=ip, port=port) LOGGER.info("ZMQ Actors constructed") # Note these sockets must be set up for testing # this is not needed for running in improv diff --git a/test/test_link.py b/test/test_link.py index 41ffbcfb4..eec9a521e 100644 --- a/test/test_link.py +++ b/test/test_link.py @@ -7,36 +7,9 @@ from improv.actor import Actor -from improv.store import StoreInterface from improv.link import Link -@pytest.fixture() -def setup_store(): - """Fixture to set up the store subprocess with 10 mb. - - This fixture runs a subprocess that instantiates the store with a - memory of 10 megabytes. It specifies that "/tmp/store/" is the - location of the store socket. - - Yields: - store: An instance of the store. - - TODO: - Figure out the scope. - """ - - p = subprocess.Popen( - ["plasma_store", "-s", "/tmp/store", "-m", str(10000000)], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) - store = StoreInterface(store_loc="/tmp/store") - yield store - p.kill() - p.wait() - - def init_actors(n=1): """Function to return n unique actors. @@ -50,17 +23,16 @@ def init_actors(n=1): return [Actor("test " + str(i), "/tmp/store", links={}) for i in range(n)] -@pytest.fixture() +@pytest.fixture def example_link(setup_store): """Fixture to provide a commonly used Link object.""" - setup_store act = init_actors(2) lnk = Link("Example", act[0].name, act[1].name) yield lnk lnk = None -@pytest.fixture() +@pytest.fixture def example_actor_system(setup_store): """Fixture to provide a list of 4 connected actors.""" @@ -88,7 +60,7 @@ def example_actor_system(setup_store): acts = None -@pytest.fixture() +@pytest.fixture def _kill_pytest_processes(): """Kills all processes with "pytest" in their name. @@ -242,7 +214,7 @@ def test_put_nowait(example_link): assert t_net < 0.005 # 5 ms -@pytest.mark.asyncio() +@pytest.mark.asyncio async def test_put_async_success(example_link): """Tests if put_async returns None. @@ -256,7 +228,7 @@ async def test_put_async_success(example_link): assert res is None -@pytest.mark.asyncio() +@pytest.mark.asyncio async def test_put_async_multiple(example_link): """Tests if async putting multiple objects preserves their order.""" @@ -273,7 +245,7 @@ async def test_put_async_multiple(example_link): assert messages_out == messages -@pytest.mark.asyncio() +@pytest.mark.asyncio async def test_put_and_get_async(example_link): """Tests if async get preserves order after async put.""" @@ -290,15 +262,17 @@ async def test_put_and_get_async(example_link): assert messages_out == messages -def test_put_overflow(setup_store, caplog): +@pytest.mark.skip( + reason="This test needs additional work to cause an overflow in the datastore." +) +def test_put_overflow(setup_store, server_port_num, caplog): """Tests if putting too large of an object raises an error.""" p = subprocess.Popen( - ["plasma_store", "-s", "/tmp/store", "-m", str(1000)], + ["redis-server", "--port", str(server_port_num), "--maxmemory", str(1000)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) - StoreInterface(store_loc="/tmp/store") acts = init_actors(2) lnk = Link("L1", acts[0], acts[1]) @@ -309,7 +283,6 @@ def test_put_overflow(setup_store, caplog): p.kill() p.wait() - setup_store # restore the 10 mb store if caplog.records: for record in caplog.records: @@ -397,7 +370,7 @@ def test_get_nowait_empty(example_link): pytest.fail("the queue is not empty") -@pytest.mark.asyncio() +@pytest.mark.asyncio async def test_get_async_success(example_link): """Tests if async_get gets the correct element from the queue.""" @@ -408,7 +381,7 @@ async def test_get_async_success(example_link): assert res == "message" -@pytest.mark.asyncio() +@pytest.mark.asyncio async def test_get_async_empty(example_link): """Tests if get_async times out given an empty queue. @@ -442,7 +415,7 @@ def test_cancel_join_thread(example_link): @pytest.mark.skip(reason="unfinished") -@pytest.mark.asyncio() +@pytest.mark.asyncio async def test_join_thread(example_link): """Tests join_thread. This test is unfinished @@ -456,7 +429,7 @@ async def test_join_thread(example_link): assert True -@pytest.mark.asyncio() +@pytest.mark.asyncio async def test_multi_actor_system(example_actor_system, setup_store): """Tests if async puts/gets with many actors have good messages.""" diff --git a/test/test_nexus.py b/test/test_nexus.py index 84f3759f5..e1a0e5f67 100644 --- a/test/test_nexus.py +++ b/test/test_nexus.py @@ -1,3 +1,5 @@ +import glob +import shutil import time import os import pytest @@ -9,14 +11,13 @@ from improv.nexus import Nexus from improv.store import StoreInterface - # from improv.actor import Actor # from improv.store import StoreInterface SERVER_COUNTER = 0 -@pytest.fixture() +@pytest.fixture def ports(): global SERVER_COUNTER CONTROL_PORT = 5555 @@ -30,7 +31,7 @@ def ports(): SERVER_COUNTER += 3 -@pytest.fixture() +@pytest.fixture def setdir(): prev = os.getcwd() os.chdir(os.path.dirname(__file__) + "/configs") @@ -38,12 +39,12 @@ def setdir(): os.chdir(prev) -@pytest.fixture() +@pytest.fixture def sample_nex(setdir, ports): nex = Nexus("test") nex.createNexus( file="good_config.yaml", - store_size=4000, + store_size=40000000, control_port=ports[0], output_port=ports[1], ) @@ -80,11 +81,16 @@ def test_init(setdir): assert str(nex) == "test" -def test_createNexus(setdir, ports): +@pytest.mark.parametrize( + "cfg_name", + [ + "good_config.yaml", + "good_config_plasma.yaml", + ], +) +def test_createNexus(setdir, ports, cfg_name): nex = Nexus("test") - nex.createNexus( - file="good_config.yaml", control_port=ports[0], output_port=ports[1] - ) + nex.createNexus(file=cfg_name, control_port=ports[0], output_port=ports[1]) assert list(nex.comm_queues.keys()) == [ "GUI_comm", "Acquirer_comm", @@ -128,7 +134,6 @@ def test_argument_config_precedence(setdir, ports): control_port=ports[0], output_port=ports[1], store_size=11_000_000, - use_hdd=True, use_watcher=True, ) cfg = nex.config.settings @@ -136,7 +141,6 @@ def test_argument_config_precedence(setdir, ports): assert cfg["control_port"] == ports[0] assert cfg["output_port"] == ports[1] assert cfg["store_size"] == 20_000_000 - assert not cfg["use_hdd"] assert not cfg["use_watcher"] @@ -210,7 +214,14 @@ def test_config_construction(cfg_name, actor_list, link_list, setdir, ports): assert True -def test_single_actor(setdir, ports): +@pytest.mark.parametrize( + "cfg_name", + [ + "single_actor.yaml", + "single_actor_plasma.yaml", + ], +) +def test_single_actor(setdir, ports, cfg_name): nex = Nexus("test") with pytest.raises(AttributeError): nex.createNexus( @@ -271,7 +282,7 @@ def test_queue_message(setdir, sample_nex): assert True -@pytest.mark.asyncio() +@pytest.mark.asyncio @pytest.mark.skip(reason="This test is unfinished.") async def test_queue_readin(sample_nex, caplog): nex = sample_nex @@ -315,9 +326,9 @@ def test_usehdd_False(): assert True -def test_startstore(caplog, set_store_loc): +def test_startstore(caplog): nex = Nexus("test") - nex._startStoreInterface(10000) # 10 kb store + nex._startStoreInterface(10000000) # 10 MB store assert any( "StoreInterface start successful" in record.msg for record in caplog.records @@ -346,6 +357,238 @@ def test_closestore(caplog): assert True +def test_specified_free_port(caplog, setdir, ports): + nex = Nexus("test") + nex.createNexus( + file="minimal_with_fixed_redis_port.yaml", + store_size=10000000, + control_port=ports[0], + output_port=ports[1], + ) + + store = StoreInterface(server_port_num=6378) + store.connect_to_server() + key = store.put("port 6378") + assert store.get(key) == "port 6378" + + assert any( + "Successfully connected to redis datastore on port 6378" in record.msg + for record in caplog.records + ) + + nex.destroyNexus() + + assert any( + "StoreInterface start successful on port 6378" in record.msg + for record in caplog.records + ) + + +def test_specified_busy_port(caplog, setdir, ports, setup_store): + nex = Nexus("test") + with pytest.raises(Exception, match="Could not start Redis on specified port."): + nex.createNexus( + file="minimal_with_fixed_default_redis_port.yaml", + store_size=10000000, + control_port=ports[0], + output_port=ports[1], + ) + + nex.destroyNexus() + + assert any( + "Could not start Redis on specified port number." in record.msg + for record in caplog.records + ) + + +def test_unspecified_port_default_free(caplog, setdir, ports): + nex = Nexus("test") + nex.createNexus( + file="minimal.yaml", + store_size=10000000, + control_port=ports[0], + output_port=ports[1], + ) + + nex.destroyNexus() + + assert any( + "StoreInterface start successful on port 6379" in record.msg + for record in caplog.records + ) + + +def test_unspecified_port_default_busy(caplog, setdir, ports, setup_store): + nex = Nexus("test") + nex.createNexus( + file="minimal.yaml", + store_size=10000000, + control_port=ports[0], + output_port=ports[1], + ) + + nex.destroyNexus() + assert any( + "StoreInterface start successful on port 6380" in record.msg + for record in caplog.records + ) + + +def test_no_aof_dir_by_default(caplog, setdir, ports): + nex = Nexus("test") + nex.createNexus( + file="minimal.yaml", + store_size=10000000, + control_port=ports[0], + output_port=ports[1], + ) + + nex.destroyNexus() + + assert "appendonlydir" not in os.listdir(".") + assert all(["improv_persistence_" not in name for name in os.listdir(".")]) + + +def test_default_aof_dir_if_none_specified(caplog, setdir, ports, server_port_num): + nex = Nexus("test") + nex.createNexus( + file="minimal_with_redis_saving.yaml", + store_size=10000000, + control_port=ports[0], + output_port=ports[1], + ) + + store = StoreInterface(server_port_num=server_port_num) + store.put(1) + + time.sleep(3) + + nex.destroyNexus() + + assert "appendonlydir" in os.listdir(".") + + if "appendonlydir" in os.listdir("."): + shutil.rmtree("appendonlydir") + else: + logging.info("didn't find dbfilename") + + logging.info("exited test") + + +def test_specify_static_aof_dir(caplog, setdir, ports, server_port_num): + nex = Nexus("test") + nex.createNexus( + file="minimal_with_custom_aof_dirname.yaml", + store_size=10000000, + control_port=ports[0], + output_port=ports[1], + ) + + store = StoreInterface(server_port_num=server_port_num) + store.put(1) + + time.sleep(3) + + nex.destroyNexus() + + assert "custom_aof_dirname" in os.listdir(".") + + if "custom_aof_dirname" in os.listdir("."): + shutil.rmtree("custom_aof_dirname") + else: + logging.info("didn't find dbfilename") + + logging.info("exited test") + + +def test_use_ephemeral_aof_dir(caplog, setdir, ports, server_port_num): + nex = Nexus("test") + nex.createNexus( + file="minimal_with_ephemeral_aof_dirname.yaml", + store_size=10000000, + control_port=ports[0], + output_port=ports[1], + ) + + store = StoreInterface(server_port_num=server_port_num) + store.put(1) + + time.sleep(3) + + nex.destroyNexus() + + assert any(["improv_persistence_" in name for name in os.listdir(".")]) + + [shutil.rmtree(db_filename) for db_filename in glob.glob("improv_persistence_*")] + + logging.info("completed ephemeral db test") + + +def test_save_no_schedule(caplog, setdir, ports, server_port_num): + nex = Nexus("test") + nex.createNexus( + file="minimal_with_no_schedule_saving.yaml", + store_size=10000000, + control_port=ports[0], + output_port=ports[1], + ) + + store = StoreInterface(server_port_num=server_port_num) + + fsync_schedule = store.client.config_get("appendfsync") + + nex.destroyNexus() + + assert "appendonlydir" in os.listdir(".") + shutil.rmtree("appendonlydir") + + assert fsync_schedule["appendfsync"] == "no" + + +def test_save_every_second(caplog, setdir, ports, server_port_num): + nex = Nexus("test") + nex.createNexus( + file="minimal_with_every_second_saving.yaml", + store_size=10000000, + control_port=ports[0], + output_port=ports[1], + ) + + store = StoreInterface(server_port_num=server_port_num) + + fsync_schedule = store.client.config_get("appendfsync") + + nex.destroyNexus() + + assert "appendonlydir" in os.listdir(".") + shutil.rmtree("appendonlydir") + + assert fsync_schedule["appendfsync"] == "everysec" + + +def test_save_every_write(caplog, setdir, ports, server_port_num): + nex = Nexus("test") + nex.createNexus( + file="minimal_with_every_write_saving.yaml", + store_size=10000000, + control_port=ports[0], + output_port=ports[1], + ) + + store = StoreInterface(server_port_num=server_port_num) + + fsync_schedule = store.client.config_get("appendfsync") + + nex.destroyNexus() + + assert "appendonlydir" in os.listdir(".") + shutil.rmtree("appendonlydir") + + assert fsync_schedule["appendfsync"] == "always" + + +@pytest.mark.skip(reason="Nexus no longer deletes files on shutdown. Nothing to test.") def test_store_already_deleted_issues_warning(caplog): nex = Nexus("test") nex._startStoreInterface(10000) @@ -384,6 +627,9 @@ def test_actor_sub(setdir, capsys, monkeypatch, ports): assert True +@pytest.mark.skip( + reason="skipping to prevent issues with orphaned stores. TODO fix this" +) def test_sigint_exits_cleanly(ports, tmp_path): server_opts = [ "improv", diff --git a/test/test_store_with_errors.py b/test/test_store_with_errors.py index d2951398d..b95781a0a 100644 --- a/test/test_store_with_errors.py +++ b/test/test_store_with_errors.py @@ -1,24 +1,21 @@ import pytest +from pyarrow import plasma -# import time -from improv.store import StoreInterface +from improv.store import StoreInterface, RedisStoreInterface, PlasmaStoreInterface -# from multiprocessing import Process from pyarrow._plasma import PlasmaObjectExists from scipy.sparse import csc_matrix import numpy as np -import pyarrow.plasma as plasma +import redis +import logging -# from pyarrow.lib import ArrowIOError -# from improv.store import ObjectNotFoundError -# from improv.store import CannotGetObjectError from improv.store import CannotConnectToStoreInterfaceError -# import pickle -import subprocess - WAIT_TIMEOUT = 10 +logger = logging.getLogger(__name__) +logger.setLevel(logging.DEBUG) + # TODO: add docstrings!!! # TODO: clean up syntax - consistent capitalization, function names, etc. @@ -27,42 +24,24 @@ # Separate each class as individual file - individual tests??? -# @pytest.fixture -# def store_loc(): -# store_loc = '/dev/shm' -# return store_loc - -# store_loc = '/dev/shm' +def test_connect(setup_store, server_port_num): + store = StoreInterface(server_port_num=server_port_num) + assert isinstance(store.client, redis.Redis) -@pytest.fixture() -# TODO: put in conftest.py -def setup_store(set_store_loc): - """Start the server""" - print("Setting up Plasma store.") - p = subprocess.Popen( - ["plasma_store", "-s", set_store_loc, "-m", str(10000000)], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) - # with plasma.start_plasma_store(10000000) as ps: - - yield p +def test_plasma_connect(setup_plasma_store, set_store_loc): + store = PlasmaStoreInterface(store_loc=set_store_loc) + assert isinstance(store.client, plasma.PlasmaClient) - # ''' Kill the server - # ''' - # print('Tearing down Plasma store.') - p.kill() - p.wait(WAIT_TIMEOUT) +def test_redis_connect(setup_store, server_port_num): + store = RedisStoreInterface(server_port_num=server_port_num) + assert isinstance(store.client, redis.Redis) + assert store.client.ping() -def test_connect(setup_store, set_store_loc): - store = StoreInterface(store_loc=set_store_loc) - assert isinstance(store.client, plasma.PlasmaClient) - -def test_connect_incorrect_path(setup_store, set_store_loc): +def test_connect_incorrect_path(setup_plasma_store, set_store_loc): # TODO: shorter name??? # TODO: passes, but refactor --- see comments store_loc = "asdf" @@ -73,13 +52,20 @@ def test_connect_incorrect_path(setup_store, set_store_loc): # # Check that the exception thrown is a CannotConnectToStoreInterfaceError # raise Exception('Cannot connect to store: {0}'.format(e)) with pytest.raises(CannotConnectToStoreInterfaceError) as e: - store = StoreInterface(store_loc=store_loc) + store = PlasmaStoreInterface(store_loc=store_loc) store.connect_store(store_loc) # Check that the exception thrown is a CannotConnectToStoreInterfaceError assert e.value.message == "Cannot connect to store at {}".format(str(store_loc)) -def test_connect_none_path(setup_store): +def test_redis_connect_wrong_port(setup_store, server_port_num): + bad_port_num = 1234 + with pytest.raises(CannotConnectToStoreInterfaceError) as e: + RedisStoreInterface(server_port_num=bad_port_num) + assert e.value.message == "Cannot connect to store at {}".format(str(bad_port_num)) + + +def test_connect_none_path(setup_plasma_store): # BUT default should be store_loc = '/tmp/store' if not entered? store_loc = None # Handle exception thrown - assert name == 'CannotConnectToStoreInterfaceError' @@ -93,7 +79,7 @@ def test_connect_none_path(setup_store): # Check that the exception thrown is a CannotConnectToStoreInterfaceError # raise Exception('Cannot connect to store: {0}'.format(e)) with pytest.raises(CannotConnectToStoreInterfaceError) as e: - store = StoreInterface(store_loc=store_loc) + store = PlasmaStoreInterface(store_loc=store_loc) store.connect_store(store_loc) # Check that the exception thrown is a CannotConnectToStoreInterfaceError assert e.value.message == "Cannot connect to store at {}".format(str(store_loc)) @@ -105,65 +91,34 @@ def test_connect_none_path(setup_store): # TODO: @pytest.parameterize...store.get and store.getID for diff datatypes, # pickleable and not, etc. # Check raises...CannotGetObjectError (object never stored) -def test_init_empty(setup_store, set_store_loc): - store = StoreInterface(store_loc=set_store_loc) - assert store.get_all() == {} +def test_init_empty(setup_store, server_port_num): + store = StoreInterface(server_port_num=server_port_num) + # logger.info(store.client.config_get()) + assert store.get_all() == [] -# class StoreInterfaceGetID(self): -# TODO: -# Check both hdd_only=False/True -# Check isInstance type, isInstance bytes, else -# Check in disk - pytest.raises(ObjectNotFoundError) -# Decide to test_getList and test_get_all +def test_plasma_init_empty(setup_plasma_store, set_store_loc): + store = PlasmaStoreInterface(store_loc=set_store_loc) + assert store.get_all() == {} -# def test_is_picklable(self): -# Test if obj to put is picklable - if not raise error, handle/suggest how to fix -# TODO: TEST BELOW: -# except PlasmaObjectExists: -# logger.error('Object already exists. Meant to call replace?') -# except ArrowIOError as e: -# logger.error('Could not store object '+ \ -# object_name+': {} {}'.format(type(e).__name__, e)) -# logger.info('Refreshing connection and continuing') -# self.reset() -# except Exception as e: -# logger.error('Could not store object '+ \ -# object_name+': {} {}'.format(type(e).__name__, e)) +def test_is_csc_matrix_and_put(setup_store, server_port_num): + mat = csc_matrix((3, 4), dtype=np.int8) + store = StoreInterface(server_port_num=server_port_num) + x = store.put(mat) + assert isinstance(store.get(x), csc_matrix) -def test_is_csc_matrix_and_put(setup_store, set_store_loc): +def test_plasma_is_csc_matrix_and_put(setup_plasma_store, set_store_loc): mat = csc_matrix((3, 4), dtype=np.int8) - store = StoreInterface(store_loc=set_store_loc) + store = PlasmaStoreInterface(store_loc=set_store_loc) x = store.put(mat, "matrix") assert isinstance(store.getID(x), csc_matrix) -# FAILED - ObjectNotFoundError NOT RAISED? -# def test_not_put(setup_store): -# store_loc = '/tmp/store' -# store = StoreInterface(store_loc) -# with pytest.raises(ObjectNotFoundError) as e: -# obj_id = store.getID(store.random_ObjectID(1)) -# # Check that the exception thrown is a ObjectNotFoundError -# assert e.value.message == 'Cannnot find object with ID/name "{}"'.format(obj_id) - -# FAILED - AssertionError...looks at LMDBStoreInterface in story.py -# assert name is not None? -# def test_use_hdd(setup_store): -# store_loc = '/tmp/store' -# store = StoreInterface(store_loc, use_lmdb=True) -# lmdb_store = store.lmdb_store -# lmdb_store.put(1, 'one') -# assert lmdb_store.getID('one', hdd_only=True) == 1 - -# class StoreInterfaceGetListandAll(StoreInterfaceDependentTestCase): - - -@pytest.mark.skip() -def test_get_list_and_all(setup_store, set_store_loc): - store = StoreInterface(store_loc=set_store_loc) +@pytest.mark.skip +def test_get_list_and_all(setup_store, server_port_num): + store = StoreInterface(server_port_num=server_port_num) # id = store.put(1, "one") # id2 = store.put(2, "two") # id3 = store.put(3, "three") @@ -171,33 +126,34 @@ def test_get_list_and_all(setup_store, set_store_loc): assert [1, 2, 3] == store.get_all() -# class StoreInterface_ReleaseReset(StoreInterfaceDependentTestCase): - -# FAILED - DID NOT RAISE