Copying Objects in Batches (SDK for C)

If you have any questions during development, post them on the Issues page of GitHub.

Sample Code

This example copies objects from an OBS bucket to a destination bucket in batches. You can list objects whose names contain the specified prefix and copy them to the destination bucket in batches based on the specified concurrency. The destination keys are generated by replacing the original prefixes. Cross-bucket replication and intra-bucket backup are supported. This method is suitable for scenarios such as data migration and backup.

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include "eSDKOBS.h"
#define MAX_KEY_LEN     1024
#define MAX_TASK        10000
typedef struct {
    char source_key[MAX_KEY_LEN];
    char target_key[MAX_KEY_LEN];
} copy_task;
typedef struct {
    int total;
    int success;
    int failed;
    char failed_keys[100][MAX_KEY_LEN];
    int failed_count;
    pthread_mutex_t mutex;
} batch_result;
static batch_result g_result = {0, 0, 0, {{0}}, 0, PTHREAD_MUTEX_INITIALIZER};
static copy_task *g_tasks = NULL;
static int g_task_count = 0;
static int g_task_index = 0;
static pthread_mutex_t g_task_mutex = PTHREAD_MUTEX_INITIALIZER;
static obs_status copy_object_properties_callback(const obs_response_properties *properties, void *callback_data)
{
    (void)properties;
    (void)callback_data;
    return OBS_STATUS_OK;
}
static void copy_object_complete_callback(obs_status status,
                                          const obs_error_details *error,
                                          void *callback_data)
{
    (void)error;
    copy_task *task = (copy_task *)callback_data;
    char *source_key = task ? task->source_key : "unknown";
    char *target_key = task ? task->target_key : "unknown";
    pthread_mutex_lock(&g_result.mutex);
    if (status == OBS_STATUS_OK) {
        g_result.success++;
        printf("[SUCCESS] copy %s -> %s\n", source_key, target_key);
    } else {
        g_result.failed++;
        if (g_result.failed_count < 100) {
            strncpy(g_result.failed_keys[g_result.failed_count], source_key, MAX_KEY_LEN - 1);
            g_result.failed_keys[g_result.failed_count][MAX_KEY_LEN - 1] = '\0';
            g_result.failed_count++;
        }
        printf("[FAILED] copy %s: %s\n", source_key, obs_get_status_name(status));
    }
    pthread_mutex_unlock(&g_result.mutex);
}
typedef struct {
    copy_task *tasks;
    int max_count;
    int current_count;
    size_t prefix_len;
    const char *target_prefix;
    obs_status ret_status;
} list_cb_data;
static obs_status list_objects_callback(int is_truncated, const char *next_marker,
                                         int contents_count,
                                         const obs_list_objects_content *contents,
                                         int common_prefixes_count,
                                         const char **common_prefixes,
                                         void *callback_data)
{
    (void)is_truncated;
    (void)next_marker;
    (void)common_prefixes_count;
    (void)common_prefixes;
    list_cb_data *data = (list_cb_data *)callback_data;
    for (int i = 0; i < contents_count && data->current_count < data->max_count; i++) {
        memset(&data->tasks[data->current_count], 0, sizeof(copy_task));
        strncpy(data->tasks[data->current_count].source_key, contents[i].key, MAX_KEY_LEN - 1);
        snprintf(data->tasks[data->current_count].target_key, MAX_KEY_LEN, "%s%s",
                 data->target_prefix, contents[i].key + data->prefix_len);
        data->current_count++;
    }
    return OBS_STATUS_OK;
}
static void list_complete_callback(obs_status status,
                                    const obs_error_details *error,
                                    void *callback_data)
{
    (void)error;
    list_cb_data *data = (list_cb_data *)callback_data;
    data->ret_status = status;
}
static void *worker_thread(void *arg)
{
    obs_options *options_ptr = (obs_options *)arg;
    const char *bucket_name = (const char *)options_ptr->bucket_options.bucket_name;
    while (1) {
        pthread_mutex_lock(&g_task_mutex);
        if (g_task_index >= g_task_count) {
            pthread_mutex_unlock(&g_task_mutex);
            break;
        }
        int idx = g_task_index++;
        pthread_mutex_unlock(&g_task_mutex);
        copy_task *task = &g_tasks[idx];
        obs_copy_destination_object_info dest_info;
        memset(&dest_info, 0, sizeof(dest_info));
        dest_info.destination_bucket = bucket_name;
        dest_info.destination_key = task->target_key;
        obs_response_handler handler = {
            copy_object_properties_callback,
            copy_object_complete_callback
        };
        copy_object(options_ptr, task->source_key, NULL, &dest_info,
                    1, NULL, 0, &handler, task);
    }
    return NULL;
}
int main()
{
    const char *BUCKET_NAME = "example-bucket-name";
    const char *PREFIX = "test/source/";
    const char *TARGET_PREFIX = "backup/";
    int THREAD_NUM = 8;
    obs_status status = obs_initialize(OBS_INIT_ALL);
    if (status != OBS_STATUS_OK) {
        printf("obs_initialize failed: %s\n", obs_get_status_name(status));
        return 1;
    }
    obs_options options;
    init_obs_options(&options);
    // Enter the endpoint corresponding to the bucket for host_name. CN-Hong Kong is used here as an example. Replace it with the one currently in use.
    options.bucket_options.host_name = "obs.ap-southeast-1.myhuaweicloud.com";
    options.bucket_options.bucket_name = (char *)BUCKET_NAME;
    options.bucket_options.access_key = getenv("ACCESS_KEY_ID");
    options.bucket_options.secret_access_key = getenv("SECRET_ACCESS_KEY");
    printf("\nListing objects with prefix '%s'...\n", PREFIX);
    g_tasks = (copy_task *)malloc(sizeof(copy_task) * MAX_TASK);
    if (!g_tasks) {
        printf("Failed to allocate memory\n");
        obs_deinitialize();
        return 1;
    }
    list_cb_data cb_data = {g_tasks, MAX_TASK, 0, strlen(PREFIX), TARGET_PREFIX, OBS_STATUS_BUTT};
    obs_list_objects_handler list_handler = {
        { NULL, list_complete_callback },
        list_objects_callback
    };
    list_bucket_objects(&options, PREFIX, NULL, NULL, 1000, &list_handler, &cb_data);
    if (cb_data.ret_status != OBS_STATUS_OK) {
        printf("list_bucket_objects failed: %s\n", obs_get_status_name(cb_data.ret_status));
        free(g_tasks);
        obs_deinitialize();
        return 1;
    }
    g_task_count = cb_data.current_count;
    if (g_task_count == 0) {
        printf("No objects found with prefix '%s'\n", PREFIX);
        free(g_tasks);
        obs_deinitialize();
        return 0;
    }
    printf("Found %d objects to copy\n", g_task_count);
    g_result.total = g_task_count;
    g_result.success = 0;
    g_result.failed = 0;
    g_result.failed_count = 0;
    g_task_index = 0;
    pthread_t threads[THREAD_NUM];
    for (int i = 0; i < THREAD_NUM; i++) {
        pthread_create(&threads[i], NULL, worker_thread, &options);
    }
    for (int i = 0; i < THREAD_NUM; i++) {
        pthread_join(threads[i], NULL);
    }
    printf("\n========== Batch Copy Summary ==========\n");
    printf("Total: %d\n", g_result.total);
    printf("Success: %d\n", g_result.success);
    printf("Failed: %d\n", g_result.failed);
    if (g_result.failed_count > 0) {
        printf("\nFailed objects:\n");
        for (int i = 0; i < g_result.failed_count; i++) {
            printf("  - %s\n", g_result.failed_keys[i]);
        }
    }
    printf("==========================================\n");
    free(g_tasks);
    obs_deinitialize();
    return 0;
}