diff options
| -rw-r--r-- | benchmark_mini.py | 2 | ||||
| -rw-r--r-- | benchmark_throughput.py | 2 | ||||
| -rw-r--r-- | femtoqueue.py | 10 | ||||
| -rw-r--r-- | test.py | 16 |
4 files changed, 15 insertions, 15 deletions
diff --git a/benchmark_mini.py b/benchmark_mini.py index ece0ea1..273b88b 100644 --- a/benchmark_mini.py +++ b/benchmark_mini.py @@ -5,7 +5,7 @@ from femtoqueue import FemtoQueue def benchmark_femtoqueue(num_tasks: int = 1000): tmpdir = tempfile.mkdtemp() - queue = FemtoQueue(data_dir=tmpdir, node_name="node1") + queue = FemtoQueue(data_dir=tmpdir, node_id="node1") data = b"x" * 100 # 100-byte payload diff --git a/benchmark_throughput.py b/benchmark_throughput.py index 8e6be43..b21c28a 100644 --- a/benchmark_throughput.py +++ b/benchmark_throughput.py @@ -8,7 +8,7 @@ from femtoqueue import FemtoQueue def run_benchmark(duration_seconds: int, payload_size: int = 100): print(f"Running throughput benchmark for {duration_seconds} sec. See -h for help.") tmpdir = tempfile.mkdtemp() - q = FemtoQueue(data_dir=tmpdir, node_name="node1") + q = FemtoQueue(data_dir=tmpdir, node_id="node1") payload = b"x" * payload_size total_count = 0 diff --git a/femtoqueue.py b/femtoqueue.py index b3e2412..55211c2 100644 --- a/femtoqueue.py +++ b/femtoqueue.py @@ -18,11 +18,11 @@ class FemtoQueue: def __init__( self, data_dir: str, - node_name: str, + node_id: str, timeout_stale_ms: int = 30_000, ): - assert node_name not in self.RESERVED_NAMES - self.node_name = node_name + assert node_id not in self.RESERVED_NAMES + self.node_id = node_id assert timeout_stale_ms > 0 self.timeout_stale_ms = timeout_stale_ms @@ -33,7 +33,7 @@ class FemtoQueue: self.data_dir = data_dir self.dir_creating = path.join(data_dir, "creating") self.dir_pending = path.join(data_dir, "pending") - self.dir_in_progress = path.join(data_dir, node_name) + self.dir_in_progress = path.join(data_dir, node_id) self.dir_done = path.join(data_dir, "done") self.dir_failed = path.join(data_dir, "failed") @@ -77,7 +77,7 @@ class FemtoQueue: # Skip non-directories and reserved names if not path.isdir(full_dir_path): continue - if dir_name in self.RESERVED_NAMES + [self.node_name]: + if dir_name in self.RESERVED_NAMES + [self.node_id]: continue # Check tasks in this node's in-progress directory @@ -19,7 +19,7 @@ def reset_time_mock(): class TestFemtoQueue(unittest.TestCase): def test_basic(self): dir = mkdtemp() - q = FemtoQueue(data_dir = dir, node_name = "node1") + q = FemtoQueue(data_dir = dir, node_id = "node1") # Add a JSON task some_data = json.dumps({ "foo": "bar" }) @@ -41,7 +41,7 @@ class TestFemtoQueue(unittest.TestCase): def test_aborted(self): dir = mkdtemp() - q = FemtoQueue(data_dir = dir, node_name = "node1") + q = FemtoQueue(data_dir = dir, node_id = "node1") # Add a JSON task some_data = json.dumps({ "foo": "bar" }) @@ -55,7 +55,7 @@ class TestFemtoQueue(unittest.TestCase): self.assertEqual(parsed_data["foo"], "bar") # Simulate a fault and start over - q = FemtoQueue(data_dir = dir, node_name = "node1") + q = FemtoQueue(data_dir = dir, node_id = "node1") # The task should still be assigned to this node task = q.pop() @@ -77,14 +77,14 @@ class TestFemtoQueue(unittest.TestCase): set_time_mock(0) # Node1 creates and claims the task - q1 = FemtoQueue(data_dir=dir, node_name="node1", timeout_stale_ms=timeout_stale_ms) + q1 = FemtoQueue(data_dir=dir, node_id="node1", timeout_stale_ms=timeout_stale_ms) q1.push(b"stuck") task = q1.pop() self.assertIsNotNone(task) task = cast(FemtoTask, task) # Assert that node2 can not see the task - q2 = FemtoQueue(data_dir=dir, node_name="node2", timeout_stale_ms=timeout_stale_ms) + q2 = FemtoQueue(data_dir=dir, node_id="node2", timeout_stale_ms=timeout_stale_ms) non_existant_task = q2.pop() self.assertIsNone(non_existant_task) @@ -101,7 +101,7 @@ class TestFemtoQueue(unittest.TestCase): def test_mark_task_failed(self): dir = mkdtemp() - q = FemtoQueue(data_dir=dir, node_name="node1") + q = FemtoQueue(data_dir=dir, node_id="node1") # Push and pop a task q.push(b"will fail") @@ -121,7 +121,7 @@ class TestFemtoQueue(unittest.TestCase): def test_mark_task_done(self): dir = mkdtemp() - q = FemtoQueue(data_dir=dir, node_name="node1") + q = FemtoQueue(data_dir=dir, node_id="node1") # Push and pop a task q.push(b"complete me") @@ -141,7 +141,7 @@ class TestFemtoQueue(unittest.TestCase): def test_fifo_order(self): dir = mkdtemp() - q = FemtoQueue(data_dir=dir, node_name="node1") + q = FemtoQueue(data_dir=dir, node_id="node1") num_tasks = 100 |