Uploading Objects in Batches (SDK for C)

Sample Code: Uploading an Object - File-Based

This example uploads files in a local directory to a specified OBS bucket in batches. You can recursively scan the local directory and upload the files concurrently based on the specified concurrency. The OBS object keys are generated according to the relative paths of the files. This method is suitable for scenarios such as local file batch migration and data backup uploads.

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <sys/stat.h>
#include <errno.h>
#include <dirent.h>
#include "eSDKOBS.h"
#define MAX_KEY_LEN     1024
#define MAX_PATH_LEN    256
#define MAX_TASK        10000
typedef struct {
    char key[MAX_KEY_LEN];
    char local_path[MAX_PATH_LEN];
} upload_task;
typedef struct {
    upload_task *task;
    FILE *fp;
} upload_context;
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 upload_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 put_object_properties_callback(const obs_response_properties *properties, void *callback_data)
{
    (void)properties;
    (void)callback_data;
    return OBS_STATUS_OK;
}
static int put_object_data_callback(int buffer_size, char *buffer, void *callback_data)
{
    upload_context *ctx = (upload_context *)callback_data;
    if (!ctx || !ctx->fp) return -1;
    size_t read = fread(buffer, 1, buffer_size, ctx->fp);
    return (int)read;
}
static void put_object_complete_callback(obs_status status,
                                          const obs_error_details *error,
                                          void *callback_data)
{
    (void)error;
    upload_context *ctx = (upload_context *)callback_data;
    char *key = ctx && ctx->task ? ctx->task->key : "unknown";
    pthread_mutex_lock(&g_result.mutex);
    if (status == OBS_STATUS_OK) {
        g_result.success++;
        printf("[SUCCESS] upload %s\n", key);
    } else {
        g_result.failed++;
        if (g_result.failed_count < 100) {
            strncpy(g_result.failed_keys[g_result.failed_count], key, MAX_KEY_LEN - 1);
            g_result.failed_keys[g_result.failed_count][MAX_KEY_LEN - 1] = '\0';
            g_result.failed_count++;
        }
        printf("[FAILED] upload %s: %s\n", key, obs_get_status_name(status));
    }
    pthread_mutex_unlock(&g_result.mutex);
}
static int scan_directory(const char *base_dir, const char *prefix, upload_task *tasks, int max_count)
{
    int count = 0;
    DIR *dir;
    struct dirent *entry;
    char full_path[MAX_PATH_LEN];
    struct stat st;
    dir = opendir(base_dir);
    if (!dir) {
        printf("Failed to open directory: %s\n", base_dir);
        return 0;
    }
    while ((entry = readdir(dir)) != NULL && count < max_count) {
        if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) {
            continue;
        }
        snprintf(full_path, sizeof(full_path), "%s/%s", base_dir, entry->d_name);
        if (stat(full_path, &st) != 0) {
            continue;
        }
        if (S_ISDIR(st.st_mode)) {
            int sub_count = scan_directory(full_path, prefix, tasks + count, max_count - count);
            count += sub_count;
        } else if (S_ISREG(st.st_mode)) {
            const char *relative_path = full_path + strlen(base_dir) + 1;
            snprintf(tasks[count].key, MAX_KEY_LEN, "%s%s", prefix, relative_path);
            strncpy(tasks[count].local_path, full_path, MAX_PATH_LEN - 1);
            tasks[count].local_path[MAX_PATH_LEN - 1] = '\0';
            count++;
        }
    }
    closedir(dir);
    return count;
}
static void *worker_thread(void *arg)
{
    (void)arg;
    obs_options *options_ptr = (obs_options *)arg;
    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);
        upload_task *task = &g_tasks[idx];
        FILE *fp = fopen(task->local_path, "rb");
        if (!fp) {
            pthread_mutex_lock(&g_result.mutex);
            g_result.failed++;
            if (g_result.failed_count < 100) {
                strncpy(g_result.failed_keys[g_result.failed_count], task->key, MAX_KEY_LEN - 1);
                g_result.failed_keys[g_result.failed_count][MAX_KEY_LEN - 1] = '\0';
                g_result.failed_count++;
            }
            printf("[FAILED] cannot open %s: %s\n", task->local_path, strerror(errno));
            pthread_mutex_unlock(&g_result.mutex);
            continue;
        }
        fseek(fp, 0, SEEK_END);
        long file_size = ftell(fp);
        fseek(fp, 0, SEEK_SET);
        upload_context ctx;
        ctx.task = task;
        ctx.fp = fp;
        obs_put_object_handler handler = {
            { put_object_properties_callback, put_object_complete_callback },
            put_object_data_callback,
            NULL
        };
        obs_put_properties put_properties;
        init_put_properties(&put_properties);
        put_properties.canned_acl = OBS_CANNED_ACL_PRIVATE;
        put_object(options_ptr, task->key, (uint64_t)file_size,
                   &put_properties, 0, &handler, &ctx);
        fclose(fp);
    }
    return NULL;
}
int main()
{
    const char *BUCKET_NAME = "example-bucket-name";
    const char *SOURCE_DIR = "/tmp/upload_files";
    const char *PREFIX = "archive/";
    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("\nScanning local directory '%s' ...\n", SOURCE_DIR);
    g_tasks = (upload_task *)malloc(sizeof(upload_task) * MAX_TASK);
    if (!g_tasks) {
        printf("Failed to allocate memory\n");
        obs_deinitialize();
        return 1;
    }
    g_task_count = scan_directory(SOURCE_DIR, PREFIX, g_tasks, MAX_TASK);
    if (g_task_count == 0) {
        printf("No files found in '%s'\n", SOURCE_DIR);
        free(g_tasks);
        obs_deinitialize();
        return 0;
    }
    printf("Found %d files to upload\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 Upload 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;
}