make the default io-big adaptive

bob/pipelines/config/distributed/sge_io_big.py

@@ -6,5 +6,15 @@ from bob.pipelines.distributed.sge_queues import QUEUE_IOBIG
 min_jobs = max_jobs = get_max_jobs(QUEUE_IOBIG)
 cluster = SGEMultipleQueuesCluster(min_jobs=min_jobs, sge_job_spec=QUEUE_IOBIG)
 cluster.scale(max_jobs)
-
+# Adapting to minimim 1 job to maximum 48 jobs
+# interval: Milliseconds between checks from the scheduler
+# wait_count: Number of consecutive times that a worker should be suggested for
+#             removal before we remove it.
+cluster.adapt(
+    minimum=min_jobs,
+    maximum=max_jobs,
+    wait_count=5,
+    interval=10,
+    target_duration="10s",
+)
 dask_client = Client(cluster)

bob/pipelines/config/distributed/sge_io_big_non_adaptive.py

+from dask.distributed import Client
+
+from bob.pipelines.distributed.sge import SGEMultipleQueuesCluster, get_max_jobs
+from bob.pipelines.distributed.sge_queues import QUEUE_IOBIG
+
+min_jobs = max_jobs = get_max_jobs(QUEUE_IOBIG)
+cluster = SGEMultipleQueuesCluster(min_jobs=min_jobs, sge_job_spec=QUEUE_IOBIG)
+cluster.scale(max_jobs)
+
+dask_client = Client(cluster)

setup.py

@@ -45,6 +45,7 @@ setup(
             "local-p32       = bob.pipelines.config.distributed.local_p32:dask_client",
             "sge             = bob.pipelines.config.distributed.sge_default:dask_client",
             "sge-io-big      = bob.pipelines.config.distributed.sge_io_big:dask_client",
+            "sge-io-big-non-adaptive = bob.pipelines.config.distributed.sge_io_big_non_adaptive:dask_client",
             "sge-gpu         = bob.pipelines.config.distributed.sge_gpu:dask_client",
         ],
     },