feat(deployer): add round robin node allocator

ctl/src/deployer/alloc.rs

@@ -1,6 +1,9 @@
 //! Worker allocation algorithms.
 
-use std::net::IpAddr;
+use std::{
+    net::IpAddr,
+    sync::atomic::{AtomicUsize, Ordering},
+};
 
 use proto::common::instance::InstanceId;
 use rand::seq::SliceRandom;
@@ -10,6 +13,7 @@ use crate::worker_mgr::WorkerDetails;
 
 /// Randomly allocates, using an uniform distribution, instances for the give
 /// amount of instances and the provided pool of `workers`.
+#[allow(dead_code)]
 pub fn rand_many(
     workers: &[WorkerDetails],
     instances: u32,
@@ -22,6 +26,22 @@ pub fn rand_many(
 }
 
 /// Randomly allocates a single instance from the provided pool of `workers`.
-pub fn _rand_single(workers: &[WorkerDetails]) -> (InstanceId, IpAddr) {
+#[allow(dead_code)]
+pub fn rand_single(workers: &[WorkerDetails]) -> (InstanceId, IpAddr) {
     rand_many(workers, 1).next().unwrap()
 }
+
+pub static COUNTER: AtomicUsize = AtomicUsize::new(0);
+
+#[allow(dead_code)]
+pub fn rr_alloc_many(
+    workers: &[WorkerDetails],
+    instances: u32,
+) -> impl Iterator<Item = (InstanceId, IpAddr)> + '_ {
+    (0..instances)
+        .map(move |_| {
+            let i = COUNTER.fetch_add(1, Ordering::Relaxed);
+            &workers[i % workers.len()]
+        })
+        .map(|w| (InstanceId(Uuid::now_v7()), w.addr))
+}

ctl/src/deployer/mod.rs

@@ -123,7 +123,7 @@ impl Deployer {
         if workers.is_empty() {
             bail!("no workers on cluster pool");
         }
-        let instances = alloc::rand_many(&workers, spec.concurrency);
+        let instances = alloc::rr_alloc_many(&workers, spec.concurrency);
         let deployment_id = DeploymentId(Uuid::now_v7());
         let service_id = Arc::new(spec.service_id.clone());