____ _ | __ )| | ___ __ _ | _ \| |/ _ \ / _` | | |_) | | (_) | (_| | |____/|_|\___/ \__, | |___/_ _ _/_| |__ ___ _ __| |_ _ _/_ / _ \ '_ \ / _ \ '__| | | | |/ _ \ __/ |_) | __/ | | | |_| | __/ \___|_.__/ \___|_| |_|\__,_|\___|
[2026-10-21][Daily]
This post follows the experiments regarding the auto-adaptive task granularity in the previous post. In previous experiments, we hard-coded different task granularity to perform preliminary benchmarks, but here, we investigate how to design such a mechanism. The following subjects are mentionned, feel free to skip to the results at the end:
As a quick reminder, let's see how we define tasks in StarPU:
void cpu_func(void *buffers[], void *cl_arg)
{
printf("Hello world\n");
}
struct starpu_codelet cl =
{
.cpu_funcs = { cpu_func },
.nbuffers = 0
};Then we call it:
struct starpu_task *task = starpu_task_create();
task->cl = &cl; /* Pointer to the codelet defined above */
/* starpu_task_submit will be a blocking call. If unset,
starpu_task_wait() needs to be called after submitting the task. */
task->synchronous = 1;
/* submit the task to StarPU */
starpu_task_submit(task);Ideally, we want that every aspect of our program could be launched as a task, or that the
children function be called as tasks. An obvious limitation is the signature of StarPU functions
(void)(void *buffers[], void *cl_arg).
Another limitation concerns data object such as vectors, matrices etc. Let's define a function more useful than our "hello world":
void scal_cpu_func(void *buffers[], void *cl_arg)
{
unsigned i;
float *factor = cl_arg;
/* length of the vector */
unsigned n = STARPU_VECTOR_GET_NX(buffers[0]);
/* CPU copy of the vector pointer */
float *val = (float *)STARPU_VECTOR_GET_PTR(buffers[0]);
for (i = 0; i < n; i++)
val[i] *= *factor;
}Here, a vector is retrieved using STARPU_VECTOR_GET_PTR and its dimensions with
STARPU_VECTOR_GET_NX. Internally, the macros get these informations from a starpu
structure defined as follows:
struct starpu_vector_interface
{
enum starpu_data_interface_id id; /**< Identifier of the interface */
uintptr_t ptr; /**< local pointer of the vector */
uintptr_t dev_handle; /**< device handle of the vector. */
size_t offset; /**< offset in the vector */
size_t nx; /**< number of elements on the x-axis of the vector */
size_t elemsize; /**< size of the elements of the vector */
size_t slice_base; /**< vector slice base, used by the StarPU OpenMP runtime support */
size_t allocsize; /**< size actually currently allocated */
};This structure interface is a representation of the data for a particular task on a specific device. For example, on a GPU, or specific CPU core, the pointer and/or offset/elemsize could be different.
The interface should clearly be differentiated from the StarPU handles. The handles are defined in the main thread, and are used to determine coherency and their usage in the tasks. An handle would be defined as:
float vector[NX];
starpu_data_handle_t vector_handle;
starpu_vector_data_register(&vector_handle, STARPU_MAIN_RAM, (uintptr_t)vector, NX, sizeof(vector[0]));It is of course a limitation: in my idea, a function could be either executed as a StarPU task, or executed normally as a regular C function inside a worker. If we define a new data inside a worker, we don't need (and cannot) to instanciate a StarPU handle. We could just directly use the raw data.
The same applies for data partitioning: when accessed from tasks, we simply use the children handles, but when invoked inside a worker (i.e. we are already in a task), we can directly access the data "by hand". Let me show you an example, here is how I partition a matrix in DAHL using StarPU data filters:
dahl_partition_vector dahl_matrix_partition_along_y(dahl_matrix const matrix)
{
size_t const nparts = dahl_matrix_get_shape(matrix).ny;
struct starpu_data_filter f =
{
.filter_func = starpu_matrix_filter_pick_vector_y,
.nchildren = nparts,
.get_child_ops = starpu_matrix_filter_pick_vector_child_ops
};
dahl_arena* origin = starpu_data_get_user_data(matrix);
return dahl_partition_init(matrix, nparts, &f, origin);
}
With the dahl partition looking like this:
typedef struct
{
// Handle of the parent's data structure which have been partitioned
starpu_data_handle_t parent;
// Number of children
size_t nparts;
// List of the children handles
starpu_data_handle_t* children;
} dahl_partition;Then, we can iterate over the children to submit a task. However, if we are in on a
worker thread, we don't have access to the handles, so we should be able to access sub-data "by
hand". Here is a little code sample that would manage this:
void* block_get_matrix_z(void* blk, size_t z)
{
if (!ctx_on_worker)
{
dahl_partition* part = (dahl_partition*)starpu_data_get_user_data(blk);
return part->children[z];
}
else
{
struct starpu_block_interface* interface = blk;
struct starpu_matrix_interface* mat = malloc(sizeof(struct starpu_matrix_interface));
mat->nx = interface->nx;
mat->ny = interface->ny;
mat->ld = interface->ldy;
mat->ptr = (uintptr_t)&((dahl_fp*)interface->ptr)[interface->ldz * z];
mat->elemsize = sizeof(dahl_fp);
mat->id = STARPU_MATRIX_INTERFACE_ID;
return mat;
}
}We'll get back to the condition (!ctx_on_worker) later, but keep in mind that it is
used to determine wether the call is from the main thread, or a worker. When called from the main
thread, we get the partition object and get the correct starpu handle with children[z].
When running from a worker, it creates a struct starpu_matrix_interface* with the
correct values.
Here a huge limitation is that the data should be contiguous. Indeed, we are able to get the
pointer to the children using
&((dahl_fp*)interface->ptr)[interface->ldz * z]; because matrices in the
block are initialized contiguously. If we wanted a different data partitioning, e.g. partitioning
over the X axis, we would have to handle that ourselves, because starpu won't be able to do that in
the context of a worker execution.
To summarize, here are the different limitations I have identified yet:
(void)(void** buffers, void* cl_arg)To implementation auto-adaptive task granularity, I think we need to distinguish 3 execution contexts:
Here is an example of an execution tree:
Let's detail particular execution path of this tree:
A->B->C, no function is invoked as a task. Each time, we pass a handle and
the main thread does the logic. Only C->D and C->E does invoke tasks,
thus D and E will run in parallel (if no data dependency of course). These
tasks are "leaves", i.e. the smallest possible tasks, so nothing can be expanded further.A->B->F does invoke a task earlier, at F. However, G
and H are children of F, so they will be invoked "inlined", meaning they
will be executed one by one, synchronously, on the same worker thread as task F.A->I, I is invoked as a single task, and all
children function are called inlined on the same worker thread as I.
With such a mechanism, we could dynamically scale the granularity of our tasks easily. A high
granularity would be expanding at the leaves, like for D, E. A medium granularity would
be F, as the task will do and G and H job. A low granularity
would be I, as the task will perform the job from 6 different functions.
The most extreme example would be no granularity at all: A is launched as a single
task and does all the job on one thread. This could make sense if we had such a small workload that
there is not point in parallelizing it (energy or runtime wise).
As we discussed, there are many limitations and challenges that we need to overcome to implement this. Here are my first steps into implementing such a mechanism.
The first issue is that I have to pass either a handle, or an interface depending on the execution context. My first idea was to implement a tagged union:
typedef struct
{
bool is_handle;
union {
starpu_data_handle_t handle; /* StarPU handle */
void *interface; /* Data Interface */
} data;
} dahl_handle;Thus I can write things like this:
void dahl_convolution_filters(dahl_blk sample, dahl_tsr filters, dahl_blk output)
{
for (size_t f = 0; f < NUM_FILTERS; f++)
{
launch_convolution(
sample,
tensor_get_block_t(filters, f),
block_get_matrix_z(output, f)
);
}
}Here the function has no clue if these are handles or interfaces, but it does not matter. Only
functions tensor_get_block_t and block_get_matrix_z will check the actual
type, to return either a sub-handle, either a sub-interface. Then we can implement the expand logic
on top of it:
void real_launch_convolution_filters(dahl_blk sample, dahl_tsr filters, dahl_blk output)
{
if (sample.is_handle && GRANULARITY == 1)
{
assert(sample.is_handle && filters.is_handle && output.is_handle);
// Launch the function as a task
int ret = starpu_task_insert(&my_cl_convolution_filters,
STARPU_R, sample.data.handle,
STARPU_R, filters.data.handle,
STARPU_W, output.data.handle, 0);
STARPU_CHECK_RETURN_VALUE(ret, "couldn't insert task");
}
else
{
// Here, sample could either be inlined (thus passing interfaces), or maybe we have handles
// but the GRANULARITY consraints haven't been respected (thus it passes handles).
dahl_convolution_filters(sample, filters, output);
}
}Here the expansion logic is a single if/else: If the objects are handles AND the granularity condition is met, we can expand the function as a task. Else, we simply call the function in a regular way. In the else branch, we could either have handle objects, if the granularity condition was not met, or interfaces if the current function is already in a task.
The last remaining bit is to provide a wrapper to call dahl_convolution_filters as a
StarPU task. As we've seen, tasks should have (void)(void**,void*), which is not the
case here. So we can implement a small wrapper:
// Small adapter to wrap interfaces into a dahl_handle object
#define MAKE_HANDLE(INTERFACE) (dahl_handle) { .is_handle = false, .data.interface = (INTERFACE) }
void launch_convolution_filters(void* buffer[], void* cl_arg)
{
// Actually call the function inside the task
dahl_convolution_filters(
MAKE_HANDLE(STARPU_BLOCK_GET(buffer[0])),
MAKE_HANDLE(STARPU_TENSOR_GET(buffer[1])),
MAKE_HANDLE(STARPU_BLOCK_GET(buffer[2]))
);
}In summary, this looks like this:
|
| Handle or Interface
v
+-------------------------------+
|real_launch_convolution_filters|----------+ Handle
+-------------------------------+ v
| +--------------------------+
| Handle or Interface |launch_convolution_filters|
v +--------------------------+
+------------------------+ |
|dahl_convolution_filters| <---------------+ Interface
+------------------------+
|
| Calling another function...
v
When real launch is called with interfaces as arguments, it can only call the real implementation "inlined" (red tasks). If called with handles, and the granularity criteria is met, it is launched with the starpu wrapper function, which calls the implementation (blue tasks). If not met, it passes the handle to the function (green tasks).
So this implementation works, but it has an obvious drawback: WE NEED 3 FUNCTIONS PER TASK. In my example I had something like this:
|
| Handle or Interface
v
+-------------------------------+
|real_launch_convolution_filters|----------+ Handle
+-------------------------------+ v
| +--------------------------+
| Handle or Interface |launch_convolution_filters|
v +--------------------------+
+------------------------+ |
|dahl_convolution_filters| <---------------+ Interface
+------------------------+
|
| For f in range(filters)
|
| Handle or Interface
v
+--------------------------+
|real_launch_convolution_2d|-----+ Handle
+--------------------------+ v
| +---------------------+
| Handle or Interface |launch_convolution_2d|
v +---------------------+
+-------------------+ |
|dahl_convolution_2d| <----------+ Interface
+-------------------+
This is a leaf, stopping here
While this issue could be solved using macros (I did), it makes the code barely readable and very hard to maintain. Let's investigate how to simplify this pattern.
In fact, using the tagged union dahl_handle introduces a lot of useless boilerplate.
This is why I need to have many wrappers to transform interface into dahl_handle. This
also complicates things for the launchers. In fact, the tagged union indicates an information that
can be guessed by the scheduling context: if we are on the main thread (green tasks), we always
manipulate handles, on worker threads, we deal with interfaces (red and blue tasks). So instead of
storing this information in data objects, we can simply use a thread_local variable
:
// Defined globally, indicates if current thread is a worker.
thread_local bool ctx_on_worker = true;
// At the beginning of main, we simply turn it to false. Only the main thread will be false.
int main(int argc, char** argv)
{
dahl_init();
ctx_on_worker = false;
// ...
}
Now, we can also refactor function signatures. Because StarPU wants
(void)(void**,void*), we won't fight against it, but rather utilize the advantage of
using void** for the buffers. The convolution filters becomes:
// Buffers: image sample, kernel, output buffer
// cl_arg: none
void dahl_convolution_filters(void* buffers[], void* cl_arg)
{
for (size_t f = 0; f < NUM_FILTERS; f++)
{
void* sub_buffers[3] = {
buffers[0],
tensor_get_block_t(buffers[1], f),
block_get_matrix_z(buffers[2], f)
};
dahl_invoke(sub_buffers, cl_arg, 3, 0, &my_cl_cconvolution_2d);
}
}Like in a regular task, we know in advance how many buffers we expect (it should be well
documented of course). However, here it doesn't matter if the buffers are pointers to interfaces or
handles. block_get_matrix_z however does, and will be able to know thanks to
ctx_on_worker. Then, we can call dahl_invoke which contains the expansion
logic:
// Invoke (either as a task or inlined) a function cl.
// Pass to it buffers and cl_arg.
// You must provide the number of buffers and the granularity level for the expansion condition.
void dahl_invoke(
void* buffers[], void* cl_arg,
size_t nbuffers, size_t granularity_level,
struct starpu_codelet* cl)
{
assert(nbuffers > 0);
// If not on worker thread, and expansion condition is met
if (!ctx_on_worker && granularity_level == GRANULARITY)
{
struct starpu_task *task = starpu_task_create();
task->cl = cl;
// Assign handles
for (size_t i = 0; i < nbuffers; i++)
task->handles[i] = buffers[i];
// --> Sumbit the function as a task <--
int ret = starpu_task_submit(task);
STARPU_CHECK_RETURN_VALUE(ret, "dahl_task_insert");
}
// Else, we are on worker thread, or expansion condition is not met
else
{
// --> Simply call the function regularly, passing handles or interfaces <--
cl->cpu_funcs[0](buffers, cl_arg);
}
}This way the convolution itself is still intact, no further modification is required. We simply
add assert(ctx_on_worker) as this task is a leaf: no further task can be expanded.
void dahl_cconvolution_2d(void *buffers[], void* cl_arg)
{
assert(ctx_on_worker);
auto in = STARPU_BLOCK_GET(buffers[0]);
auto in_p = (dahl_fp const*)in.ptr;
// Kernel block
auto ker = STARPU_BLOCK_GET(buffers[1]);
auto ker_p = (dahl_fp const*)ker.ptr;
// Output matrix
auto out = STARPU_MATRIX_GET(buffers[2]);
auto out_p = (dahl_fp*)out.ptr;
// ... implementation ...
}In conclusion, this implementation is strictly better: we don't need to change current tasks
signatures and argument loading (STARPU_BLOCK_GET and all...), we don't need wrappers,
and we only need a single invoke function, as it handles arbitrary number of buffers/arguments.
Let's see this implementation in action. For now I have only implemented the two tasks we've discussed: convolution_filters(block, tensor, tensor), and convolution_2d(block, block, matrix). The first one launches convolution_2d over the filter dimensions. We only have two possible granularities.
Using granularity 0, we obtain tasks lasting around 18~50ms:
Using granularity 1, we obtain tasks lasting around 76~127ms:
Here, we have 4 filters, so in granularity 1, each task does the job of four tasks in granularity 0. Theoretically, if we multiply the run-time of granularity 0 by 4, we should obtain similar values: 18x4=72ms, compared to 76ms, makes sense. 50x4=200ms compared to 127ms. So here it seems that grouping using granularity 1 gives better runtime in the worst case. This could clearly be related to cache effects: in granularity 1, all convolutions for a single image are performed on the same worker. Using granularity 0, we (potentially) launch convolutions for the same image on multiple threads.
(As a side note, the first tasks on the figures are simply image loading tasks. They were not implemented at all with auto-adaptive granularity and don't really matter for these experiments.)