batch jobs in Longtail_ChangeVersion2 (#231)

- **NEW API** `Longtail_RunJobsBatched` runs jobs in batched mode to handle a job count larger than Longtail_JobAPI::GetMaxBatchCount() - **FIXED** Fixed memory leaks in command tool - **FIXED** `Longtail_ChangeVersion2()` can now handle workloads with a block count larger than 65535 - **FIXED** Bikeshed JobAPI implementation does efficient wait when task queue is full - **FIXED** Bikeshed JobAPI::CreateJobs implementation now properly drains both task channels when task queue is full
DanEngelbrecht · Jan 9, 2024 · 1eaf04b · 1eaf04b
1 parent d4eb67b
commit 1eaf04b
Show file tree

Hide file tree

Showing 4 changed files with 218 additions and 150 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -8,7 +8,11 @@
 - **ADDED** memtracer now tracks allocations in stb_ds
 - **ADDED** memtracer now tracks allocations in zstd
 - **NEW API** `Longtail_CompareAndSwap` compare and swap with platform implementations
+- **NEW API** `Longtail_RunJobsBatched` runs jobs in batched mode to handle a job count larger than Longtail_JobAPI::GetMaxBatchCount()
 - **FIXED** Fixed memory leaks in command tool
+- **FIXED** `Longtail_ChangeVersion2()` can now handle workloads with a block count larger than 65535
+- **FIXED** Bikeshed JobAPI implementation does efficient wait when task queue is full
+- **FIXED** Bikeshed JobAPI::CreateJobs implementation now properly drains both task channels when task queue is full
 - **CHANGED** Refactored all internal usage of JobAPI `ReadyJobs` with new error handling
 
 ## 0.4.1

diff --git a/lib/bikeshed/longtail_bikeshed.c b/lib/bikeshed/longtail_bikeshed.c
@@ -386,30 +386,40 @@ static int Bikeshed_CreateJobs(
         ctxs[i] = job_wrapper;
     }
 
-    Longtail_AtomicAdd32(&bikeshed_job_group->m_PendingJobCount, (int)job_count);
-
     while (!Bikeshed_CreateTasks(bikeshed_job_api->m_Shed, job_count, funcs, ctxs, task_ids))
     {
-        if (bikeshed_job_group->m_DetectedError == 0)
+        err = bikeshed_job_group->m_DetectedError;
+        if (err)
         {
-            if (progressAPI && is_reserve_thread)
-            {
-                progressAPI->OnProgress(progressAPI,(uint32_t)bikeshed_job_group->m_ReservedJobCount, (uint32_t)bikeshed_job_group->m_JobsCompleted);
-            }
-            if (optional_cancel_api && optional_cancel_token)
+            goto on_error;
+        }
+        if (progressAPI && is_reserve_thread)
+        {
+            progressAPI->OnProgress(progressAPI,(uint32_t)bikeshed_job_group->m_ReservedJobCount, (uint32_t)bikeshed_job_group->m_JobsCompleted);
+        }
+        if (optional_cancel_api && optional_cancel_token)
+        {
+            if (optional_cancel_api->IsCancelled(optional_cancel_api, optional_cancel_token) == ECANCELED)
             {
-                if (optional_cancel_api->IsCancelled(optional_cancel_api, optional_cancel_token) == ECANCELED)
-                {
-                    Longtail_CompareAndSwap(&bikeshed_job_group->m_DetectedError, 0, ECANCELED);
-                }
+                Longtail_CompareAndSwap(&bikeshed_job_group->m_DetectedError, 0, ECANCELED);
             }
         }
-        Bikeshed_ExecuteOne(bikeshed_job_api->m_Shed, 0);
+        if (Bikeshed_ExecuteOne(bikeshed_job_api->m_Shed, 0))
+        {
+            continue;
+        }
+        if (Bikeshed_ExecuteOne(bikeshed_job_api->m_Shed, 1))
+        {
+            continue;
+        }
+        Longtail_WaitSema(bikeshed_job_api->m_ReadyCallback.m_Semaphore, 100);
     }
+
     Bikeshed_SetTasksChannel(bikeshed_job_api->m_Shed, job_count, task_ids, job_channel);
+    Longtail_AtomicAdd32(&bikeshed_job_group->m_PendingJobCount, (int)job_count);
 
-    *out_jobs = task_ids;
     err = 0;
+    *out_jobs = task_ids;
 end:
     Longtail_Free(work_mem);
     return err;
@@ -439,7 +449,15 @@ static int Bikeshed_AddDependecies(struct Longtail_JobAPI* job_api, uint32_t job
     struct BikeshedJobAPI* bikeshed_job_api = (struct BikeshedJobAPI*)job_api;
     while (!Bikeshed_AddDependencies(bikeshed_job_api->m_Shed, job_count, (Bikeshed_TaskID*)jobs, dependency_job_count, (Bikeshed_TaskID*)dependency_jobs))
     {
-        Bikeshed_ExecuteOne(bikeshed_job_api->m_Shed, 0);
+        if (Bikeshed_ExecuteOne(bikeshed_job_api->m_Shed, 0))
+        {
+            continue;
+        }
+        if (Bikeshed_ExecuteOne(bikeshed_job_api->m_Shed, 1))
+        {
+            continue;
+        }
+        Longtail_WaitSema(bikeshed_job_api->m_ReadyCallback.m_Semaphore, 100);
     }
     return 0;
 }