Feature metadata serialization#305
Open
biqar wants to merge 51 commits into
Open
Conversation
Added code to add kv-tags at scale. Also added code to get and verify tags at scale. The later test code will be used to verify the metadata checkpointing after a server restart.
…nd from file instead of using a buffer Also added version control for the bulki-based checkpointing.
Added comment, fixed formatting suggested by GitHub clang-format style checker, brought back PDC_TIMING log from the stable version.
…into feature_metadata-serialization
…into feature_metadata-serialization
Collaborator
Author
Benchmark Summary (10000 objects, 100 tags)Serial
Parallel (
|
| Metric | Previous | New (BUKLI) |
|---|---|---|
| Close Time (s) | 0.85 | 0.45 |
| Restart Time (s) | 0.77 | 0.05 |
Collaborator
Author
Member
|
@biqar a couple of things:
|
Collaborator
Author
|
Todos:
|
Collaborator
Author
Strong Scaling Performance ComparisonFixed work to 100K objects and 100 tags creation and changing the number of parallel servers. Reporting min/max/avg of Checkpoint Time (seconds)
Server Close Time (seconds)
Server Restart Time (seconds)
Observations:
Weak Scaling Performance ComparisonFixed the amount of workload per-worker. Reporting min/max/avg of Checkpoint Time (seconds)
Close Time (seconds)
Restart Time (seconds)
Observations
Issue: Unexpected Checkpoint Overhead After Restart with BULKI SerializerDescriptionThe BULKI Serializer integration introduces unexpected performance overhead in the second checkpoint cycle. Since no new data is added after the restart, both checkpoint times should be roughly equivalent. However, we observe a consistent 29–52% increase in the second checkpoint time compared to the first. Workflow
Expected BehaviorSince no new data is added after the restart, the 1st and 2nd checkpoint times should be approximately equal (as showing in the existing binary serializer based checkpoint time comparison bellow). Observed BehaviorThe 2nd checkpoint time is consistently 29–52% more expensive than the 1st under the BULKI Serializer. BULKI Serializer: Checkpoint Time Comparison (seconds)
Binary Serializer: Checkpoint Time Comparison (seconds)
|
Collaborator
|
@biqar could you time each BULKI operation so we can see what's taking so long on the strong scaling? For instance if the |
Collaborator
Author
Strong Scaling Performance ComparisonFixed work to 100K objects and 100 tags creation and changing the number of parallel servers. Reporting min/max/avg of Server Close Time (seconds)
Server Restart Time (seconds)
Weak Scaling Performance ComparisonFixed the amount of workload per-worker. Reporting min/max/avg of Close Time (seconds)
Restart Time (seconds)
Issue: Unexpected Checkpoint Overhead After Restart with BULKI SerializerDescriptionThe BULKI Serializer integration previously observed an unexpected performance overhead in the second checkpoint cycle. Since no new data is added after the restart, both checkpoint times should be roughly equivalent. However, we previously observe a consistent 29–52% increase in the second checkpoint time compared to the first. The new fix in avoiding recursive size computation have fixed the issue. Workflow
Expected BehaviorSince no new data is added after the restart, the 1st and 2nd checkpoint times should be approximately equal. Observed BehaviorThe BULKI Serializer show similar trend as the old binary serializer. BULKI Serializer: Server Close Time Comparison (seconds)
Binary Serializer: Checkpoint Time Comparison (seconds)
|
There was a problem hiding this comment.
Pull request overview
This PR integrates BULKI-based serialization into the PDC server checkpoint/restart path to persist and restore in-memory metadata (including transfer-request query state), and adds new scale-style kvtag tests plus checkpoint/restart test runners to validate the behavior.
Changes:
- Switch server checkpoint/restart from raw
fwrite/freadblobs to a BULKI-serialized checkpoint tree (containers, metadata, regions, transfer-query state). - Extend BULKI with incremental size/accounting + preallocated array helpers and add corresponding serde tests.
- Add kvtag add/verify scale tests and new checkpoint/restart runners (serial + MPI) wired into CTest.
Reviewed changes
Copilot reviewed 12 out of 13 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| src/server/pdc_server.c | Reworks checkpoint/restart implementation to build/parse a BULKI checkpoint tree and initializes transfer-query restore via BULKI. |
| src/server/pdc_server_region/pdc_server_region_transfer_metadata_query.c | Adds BULKI-based init/checkpoint support for transfer metadata query state. |
| src/server/pdc_server_region/include/pdc_server_region_transfer_metadata_query.h | Exposes new BULKI-based init/checkpoint APIs. |
| src/commons/serde/bulki/include/bulki.h | Adds capacity + incremental append/put/delete APIs for efficient bulk construction. |
| src/commons/serde/bulki/bulki.c | Implements incremental APIs, array capacity handling, and cached-size updates. |
| src/commons/serde/bulki/bulki_serde.c | Adds notes around redundant tree walks and buffer-then-fwrite serialization behavior. |
| src/commons/serde/bulki/bulki_serde_test.c | Adds tests validating incremental size accounting and round-trip correctness. |
| src/tests/tags/kvtag_add_scale.c | New scale test for creating many objects and adding tags (used in checkpoint/restart tests). |
| src/tests/tags/kvtag_get_verify_scale.c | New scale test for reading/verifying tags across many objects (used in checkpoint/restart tests). |
| src/tests/run_checkpoint_restart_test_with_prog_param.sh | New serial checkpoint/restart runner that accepts program+args strings. |
| src/tests/run_checkpoint_restart_mpi_test_with_prog_param.sh | New MPI checkpoint/restart runner that accepts program+args strings. |
| src/tests/CMakeLists.txt | Registers new executables/scripts and adds serial + MPI checkpoint/restart tests. |
| src/api/close_server.c | Minor formatting/whitespace change. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+81
to
+104
| #ifdef ENABLE_MPI | ||
| MPI_Init(&argc, &argv); | ||
| MPI_Comm_size(MPI_COMM_WORLD, &proc_num); | ||
| MPI_Comm_rank(MPI_COMM_WORLD, &my_rank); | ||
| #endif | ||
| if (argc < 3) { | ||
| if (my_rank == 0) | ||
| print_usage(argv[0]); | ||
| PGOTO_DONE(FAIL); | ||
| } | ||
| n_obj = atoi(argv[1]); | ||
| n_add_tag = atoi(argv[2]); | ||
|
|
||
| if (n_add_tag > n_obj) { | ||
| if (my_rank == 0) | ||
| LOG_ERROR("n_add_tag larger than n_obj! Exiting...\n"); | ||
| PGOTO_DONE(FAIL); | ||
| } | ||
|
|
||
| assign_work_to_rank(my_rank, proc_num, n_add_tag, &my_add_tag, &my_add_tag_s); | ||
| assign_work_to_rank(my_rank, proc_num, n_obj, &my_obj, &my_obj_s); | ||
|
|
||
| obj_1percent = my_obj / 100; | ||
| tag_1percent = my_add_tag / 100; |
Comment on lines
+141
to
+142
| if (i > 0 && i % obj_1percent == 0) { | ||
| #ifdef ENABLE_MPI |
Comment on lines
+180
to
+187
| if (i % tag_1percent == 0) { | ||
| #ifdef ENABLE_MPI | ||
| MPI_Barrier(MPI_COMM_WORLD); | ||
| percent_time = MPI_Wtime() - stime; | ||
| if (my_rank == 0) { | ||
| int current_percentage = i / tag_1percent; | ||
| int estimated_current_tag_number = n_obj / 100 * current_percentage; | ||
| double tps = estimated_current_tag_number / percent_time; |
Comment on lines
+89
to
+111
| #ifdef ENABLE_MPI | ||
| MPI_Init(&argc, &argv); | ||
| MPI_Comm_size(MPI_COMM_WORLD, &proc_num); | ||
| MPI_Comm_rank(MPI_COMM_WORLD, &my_rank); | ||
| #endif | ||
| if (argc < 3) { | ||
| if (my_rank == 0) | ||
| print_usage(argv[0]); | ||
| PGOTO_DONE(FAIL); | ||
| } | ||
| n_obj = atoi(argv[1]); | ||
| n_query = atoi(argv[2]); | ||
|
|
||
| if (n_query > n_obj) { | ||
| if (my_rank == 0) | ||
| LOG_ERROR("n_query larger than n_obj! Exiting...\n"); | ||
| PGOTO_DONE(FAIL); | ||
| } | ||
|
|
||
| assign_work_to_rank(my_rank, proc_num, n_query, &n_tag, &n_tag_s); | ||
| assign_work_to_rank(my_rank, proc_num, n_obj, &my_obj, &my_obj_s); | ||
|
|
||
| obj_1percent = my_obj / 100; |
Comment on lines
+97
to
+104
| BULKI_Entity *ndim_ent = BULKI_get(obj_bulki, BULKI_singleton_ENTITY("ndim", PDC_STRING)); | ||
| if (ndim_ent == NULL) { | ||
| LOG_ERROR("Missing ndim in checkpoint object\n"); | ||
| PDC_free(obj_pkg); | ||
| continue; | ||
| } | ||
| memcpy(&obj_pkg->ndim, ndim_ent->data, sizeof(int)); | ||
|
|
Comment on lines
+135
to
+146
| BULKI_Entity *offset_size_ent = | ||
| BULKI_get(region_bulki, BULKI_singleton_ENTITY("reg_offset_size", PDC_STRING)); | ||
| if (offset_size_ent != NULL) { | ||
| memcpy(region_pkg->reg_offset, offset_size_ent->data, | ||
| sizeof(uint64_t) * obj_pkg->ndim * 2); | ||
| } | ||
| else { | ||
| LOG_ERROR("Missing reg_offset_size in checkpoint region\n"); | ||
| PDC_free(region_pkg->reg_offset); | ||
| PDC_free(region_pkg); | ||
| continue; | ||
| } |
Add null checks after every required BULKI_get() call in PDC_Server_restart() so malformed or incomplete checkpoint files fail with a clear error instead of dereferencing a null entity pointer.
…into feature_metadata-serialization # Conflicts: # src/server/pdc_server.c
When removing the last remaining key, the function decremented numKeys and then always copied from index numKeys - 1. With numKeys equal to 0, that index underflows and can corrupt memory or crash. Skip the swap when the deleted key is already at the last index. Also reject NULL bulki or key arguments at entry.
Derive per-rank tag add/query work from each rank's object partition so object indices and tag values stay aligned when n_tag != n_obj. Initialize proc_num and my_rank in serial builds, guard progress reporting against division by zero when per-rank work is under 100, and base tag/query TPS estimates on n_add_tag/n_query instead of n_obj.
Reject ndim outside [1, DIM_MAX] before allocating region buffers, and fail restore on missing or mismatched reg_offset_size so malformed checkpoints cannot trigger oversized allocations or memcpy overflows.
…into feature_metadata-serialization
Comment on lines
+667
to
+686
| BULKI_Entity *value = NULL; | ||
| size_t last_index = bulki->numKeys - 1; | ||
| for (size_t i = 0; i < bulki->numKeys; i++) { | ||
| if (BULKI_Entity_equal(&bulki->header->keys[i], key)) { | ||
| value = &bulki->data->values[i]; | ||
| bulki->header->headerSize -= key->size; | ||
| bulki->data->dataSize -= value->size; | ||
| bulki->numKeys--; | ||
| if (i != last_index) { | ||
| memcpy(&bulki->header->keys[i], &bulki->header->keys[bulki->numKeys - 1], | ||
| sizeof(BULKI_Entity)); | ||
| memcpy(&bulki->data->values[i], &bulki->data->values[bulki->numKeys - 1], | ||
| sizeof(BULKI_Entity)); | ||
| } | ||
| break; | ||
| } | ||
| } | ||
| bulki_refresh_total_size(bulki); | ||
|
|
||
| FUNC_LEAVE(value); |
Comment on lines
+1856
to
+1865
| BULKI_Entity *range_ent = | ||
| BULKI_get(histogram, BULKI_singleton_ENTITY("range", PDC_STRING)); | ||
| region_list->region_hist->range = | ||
| (double *)PDC_malloc(sizeof(double) * (size_t)nbin * 2); | ||
| if (range_ent == NULL || range_ent->data == NULL) | ||
| PGOTO_ERROR(FAIL, "Missing histogram range in checkpoint"); | ||
| region_list->region_hist->range = | ||
| (double *)PDC_malloc(sizeof(double) * (size_t)nbin * 2); | ||
| memcpy(region_list->region_hist->range, range_ent->data, | ||
| sizeof(double) * (size_t)nbin * 2); |
Comment on lines
+1867
to
+1876
| BULKI_Entity *bin_ent = | ||
| BULKI_get(histogram, BULKI_singleton_ENTITY("bin", PDC_STRING)); | ||
| region_list->region_hist->bin = | ||
| (uint64_t *)PDC_malloc(sizeof(uint64_t) * (size_t)nbin); | ||
| if (bin_ent == NULL || bin_ent->data == NULL) | ||
| PGOTO_ERROR(FAIL, "Missing histogram bin in checkpoint"); | ||
| region_list->region_hist->bin = | ||
| (uint64_t *)PDC_malloc(sizeof(uint64_t) * (size_t)nbin); | ||
| memcpy(region_list->region_hist->bin, bin_ent->data, | ||
| sizeof(uint64_t) * (size_t)nbin); |
Comment on lines
+1964
to
+1968
| // extract obj_id | ||
| BULKI_Entity *obj_id_ent = | ||
| BULKI_get(dataserver_obj, BULKI_singleton_ENTITY("obj_id", PDC_STRING)); | ||
| memcpy(&new_obj_reg->obj_id, obj_id_ent->data, sizeof(uint64_t)); | ||
|
|
Comment on lines
+1984
to
+1987
| BULKI_Entity *region_ent = | ||
| BULKI_get(ds_region, BULKI_singleton_ENTITY("region", PDC_STRING)); | ||
| memcpy(region_list, region_ent->data, sizeof(region_list_t)); | ||
|
|
Comment on lines
+256
to
+259
| // set return value based on verification results | ||
| if (verified_fail > 0) { | ||
| ret_value = FAIL; | ||
| } |
Comment on lines
+152
to
+155
| #ifdef ENABLE_MPI | ||
| MPI_Barrier(MPI_COMM_WORLD); | ||
| stime = MPI_Wtime(); | ||
| #endif |
Comment on lines
+178
to
+185
| #ifdef ENABLE_MPI | ||
| MPI_Barrier(MPI_COMM_WORLD); | ||
| total_time = MPI_Wtime() - stime; | ||
| #endif | ||
|
|
||
| if (my_rank == 0) | ||
| LOG_INFO("Total time to create %11d objects: %7.2f , throughput %10.2f \n", n_obj, total_time, | ||
| n_obj / total_time); |
Comment on lines
+1287
to
1299
| BULKI *container_entry = BULKI_init(2); | ||
|
|
||
| hash_key = PDC_get_hash_by_name(cont_head->cont_name); | ||
| fwrite(&hash_key, sizeof(uint32_t), 1, file); | ||
| fwrite(cont_head, sizeof(pdc_cont_hash_table_entry_t), 1, file); | ||
| BULKI_put_incremental(container_entry, BULKI_singleton_ENTITY("hash_key", PDC_STRING), | ||
| BULKI_ENTITY(&hash_key, 1, PDC_UINT32, PDC_CLS_ITEM)); | ||
|
|
||
| // store the container structure as binary blob | ||
| BULKI_put_incremental( | ||
| container_entry, BULKI_singleton_ENTITY("cont_data", PDC_STRING), | ||
| BULKI_ENTITY(cont_head, sizeof(pdc_cont_hash_table_entry_t), PDC_UINT8, PDC_CLS_ARRAY)); | ||
|
|
||
| BULKI_ENTITY_append_BULKI_incremental(containers_array, container_entry); | ||
| } |
Comment on lines
+368
to
+387
| region_temp = obj_temp->regions; | ||
| while (region_temp) { | ||
| BULKI *region_bulki = BULKI_init(2); | ||
|
|
||
| BULKI_put_incremental(region_bulki, BULKI_singleton_ENTITY("data_server_id", PDC_STRING), | ||
| BULKI_ENTITY(®ion_temp->data_server_id, 1, PDC_UINT32, PDC_CLS_ITEM)); | ||
|
|
||
| BULKI_put_incremental( | ||
| region_bulki, BULKI_singleton_ENTITY("reg_offset_size", PDC_STRING), | ||
| BULKI_ENTITY(region_temp->reg_offset, obj_temp->ndim * 2, PDC_UINT64, PDC_CLS_ARRAY)); | ||
|
|
||
| BULKI_ENTITY_append_BULKI_incremental(regions_array, region_bulki); | ||
| region_temp = region_temp->next; | ||
| } | ||
|
|
||
| BULKI_put_incremental(obj_bulki, BULKI_singleton_ENTITY("regions", PDC_STRING), regions_array); | ||
|
|
||
| BULKI_ENTITY_append_BULKI_incremental(objects_array, obj_bulki); | ||
|
|
||
| obj_temp = obj_temp->next; |
…into feature_metadata-serialization
Comment on lines
+1819
to
+1823
| if (region_ent == NULL || region_ent->data == NULL) | ||
| PGOTO_ERROR(FAIL, "Missing region in checkpoint"); | ||
| memcpy(region_list, region_ent->data, sizeof(region_list_t)); | ||
|
|
||
| // extract histogram flag |
Comment on lines
+1985
to
+1989
| if (region_ent == NULL || region_ent->data == NULL) | ||
| PGOTO_ERROR(FAIL, "Missing/invalid dataserver region blob in checkpoint"); | ||
| memcpy(region_list, region_ent->data, sizeof(region_list_t)); | ||
|
|
||
| // initialize fields (similar to above) |
Comment on lines
+1803
to
+1807
| int n_region = 0; | ||
| if (regions_array != NULL && regions_array->pdc_type == PDC_BULKI) { | ||
| n_region = regions_array->count; | ||
| if (n_region < 0 || n_region > PDC_CHECKPOINT_MAX_REGION_COUNT) | ||
| PGOTO_ERROR(FAIL, "Suspicious n_region value %d from checkpoint", n_region); |
| BULKI_Entity *metadata_objs_array = | ||
| BULKI_get(hash_entry, BULKI_singleton_ENTITY("metadata_objects", PDC_STRING)); | ||
|
|
||
| if (metadata_objs_array != NULL && metadata_objs_array->pdc_type == PDC_BULKI) { |
Comment on lines
+1298
to
1299
| BULKI_ENTITY_append_BULKI_incremental(containers_array, container_entry); | ||
| } |
Comment on lines
+379
to
+381
| BULKI_ENTITY_append_BULKI_incremental(regions_array, region_bulki); | ||
| region_temp = region_temp->next; | ||
| } |
Comment on lines
+385
to
+387
| BULKI_ENTITY_append_BULKI_incremental(objects_array, obj_bulki); | ||
|
|
||
| obj_temp = obj_temp->next; |
Comment on lines
+607
to
+615
| if (existing_value != NULL) { | ||
| bulki->header->headerSize -= key->size; | ||
| bulki->data->dataSize -= existing_value->size; | ||
| memcpy(existing_value, value, sizeof(BULKI_Entity)); | ||
| bulki->header->headerSize += key->size; | ||
| bulki->data->dataSize += value->size; | ||
| bulki_refresh_total_size(bulki); | ||
| FUNC_LEAVE_VOID(); | ||
| } |
Comment on lines
877
to
880
| else if (bulk_entity->pdc_class == PDC_CLS_ITEM) { | ||
| if (bulk_entity->pdc_type == PDC_BULKI && bulk_entity->data != NULL) { | ||
| BULKI_free((BULKI *)bulk_entity->data, 0); | ||
| bulk_entity->data = NULL; |
…into feature_metadata-serialization
houjun
approved these changes
Jul 10, 2026
TheAssembler1
approved these changes
Jul 14, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Related Issues / Pull Requests
Related Issue: 282
Description
Integrate BULKI serializer for pdc server checkpointing.
What changes are proposed in this pull request?
Checklist: