// Licensed to the Apache Software Foundation (ASF) under one // or more contributor license agreements. See the NOTICE file // distributed with this work for additional information // regarding copyright ownership. The ASF licenses this file // to you under the Apache License, Version 2.0 (the // "License"); you may not use this file except in compliance // with the License. You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, // software distributed under the License is distributed on an // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY // KIND, either express or implied. See the License for the // specific language governing permissions and limitations // under the License. #include #include #include #include #include #include #include #include // ANSI color codes #define GREEN "\033[32m" #define RED "\033[31m" #define BLUE "\033[34m" #define YELLOW "\033[33m" #define CYAN "\033[36m" #define RESET "\033[0m" const int NUM_THREADS = 20; const int THREADS_PER_BUCKET = 5; // Simple global config struct { std::string username = "Administrator"; std::string password = "password"; std::vector bucket_names = {"t0", "t1", "t2", "t3"}; } g_config; // Simple thread statistics struct ThreadStats { std::atomic operations_attempted{0}; std::atomic operations_successful{0}; std::atomic operations_failed{0}; void reset() { operations_attempted = 0; operations_successful = 0; operations_failed = 0; } }; // Global statistics struct GlobalStats { ThreadStats total; std::vector per_thread_stats; GlobalStats() : per_thread_stats(NUM_THREADS) {} void aggregate_stats() { total.reset(); for (const auto& stats : per_thread_stats) { total.operations_attempted += stats.operations_attempted.load(); total.operations_successful += stats.operations_successful.load(); total.operations_failed += stats.operations_failed.load(); } } } g_stats; // Simple thread arguments struct ThreadArgs { int thread_id; int bucket_id; std::string bucket_name; brpc::CouchbaseOperations* couchbase_ops; ThreadStats* stats; }; // Simple CRUD operations on default collection void perform_crud_operations_default(brpc::CouchbaseOperations& couchbase_ops, const std::string& base_key, ThreadStats* stats) { std::string key = base_key + "_default"; std::string value = butil::string_printf( R"({"thread_id": %d, "collection": "default"})", (int)bthread_self()); stats->operations_attempted++; // UPSERT brpc::CouchbaseOperations::Result result = couchbase_ops.upsert(key, value); if (result.success) { stats->operations_successful++; } else { stats->operations_failed++; return; } stats->operations_attempted++; // GET result = couchbase_ops.get(key); if (result.success) { stats->operations_successful++; } else { stats->operations_failed++; return; } stats->operations_attempted++; // DELETE result = couchbase_ops.delete_(key); if (result.success) { stats->operations_successful++; } else { stats->operations_failed++; } } // Simple CRUD operations on col1 collection void perform_crud_operations_col1(brpc::CouchbaseOperations& couchbase_ops, const std::string& base_key, ThreadStats* stats) { std::string key = base_key + "_col1"; std::string value = butil::string_printf( R"({"thread_id": %d, "collection": "col1"})", (int)bthread_self()); stats->operations_attempted++; // UPSERT brpc::CouchbaseOperations::Result result = couchbase_ops.upsert(key, value, "col1"); if (result.success) { stats->operations_successful++; } else { stats->operations_failed++; std::cout << "UPSERT failed: " << result.error_message << std::endl; return; } stats->operations_attempted++; // GET result = couchbase_ops.get(key, "col1"); if (result.success) { stats->operations_successful++; } else { stats->operations_failed++; std::cout << "GET failed: " << result.error_message << std::endl; return; } stats->operations_attempted++; // DELETE result = couchbase_ops.delete_(key, "col1"); if (result.success) { stats->operations_successful++; } else { stats->operations_failed++; } } // Simple thread worker function void* thread_worker(void* arg) { ThreadArgs* args = static_cast(arg); std::cout << CYAN << "Thread " << args->thread_id << " starting on bucket " << args->bucket_name << RESET << std::endl; // Create CouchbaseOperations instance for this thread brpc::CouchbaseOperations couchbase_ops; // Authentication brpc::CouchbaseOperations::Result auth_result = couchbase_ops.authenticate( g_config.username, g_config.password, "127.0.0.1:11210", args->bucket_name); // for SSL authentication use below line instead // brpc::CouchbaseOperations::Result auth_result = couchbase_ops.authenticateSSL(username, password, "127.0.0.1:11210", args->bucket_name, "/path/to/cert.txt"); if (!auth_result.success) { std::cout << RED << "Thread " << args->thread_id << ": Auth failed - " << auth_result.error_message << RESET << std::endl; return NULL; } // // Select bucket // brpc::CouchbaseOperations::Result bucket_result = // couchbase_ops.selectBucket(args->bucket_name); // if (!bucket_result.success) { // std::cout << RED << "Thread " << args->thread_id // << ": Bucket selection failed - " << bucket_result.error_message // << RESET << std::endl; // return NULL; // } std::cout << GREEN << "Thread " << args->thread_id << " connected to bucket " << args->bucket_name << RESET << std::endl; // Perform operations - 10 times on default collection, 10 times on col1 // collection for (int i = 0; i < 10; ++i) { std::string base_key = butil::string_printf("thread_%d_op_%d", args->thread_id, i); // CRUD operations on default collection perform_crud_operations_default(couchbase_ops, base_key, args->stats); // CRUD operations on col1 collection perform_crud_operations_col1(couchbase_ops, base_key, args->stats); // Small delay between operations bthread_usleep(10000); // 10ms } int successful = args->stats->operations_successful.load(); int attempted = args->stats->operations_attempted.load(); int failed = args->stats->operations_failed.load(); std::cout << GREEN << "Thread " << args->thread_id << " completed: " << successful << "/" << attempted << " operations successful, " << failed << " failed" << RESET << std::endl; return NULL; } void* shared_object_thread_worker(void *arg){ ThreadArgs* shared_args = static_cast(arg); brpc::CouchbaseOperations* shared_couchbase_ops = shared_args->couchbase_ops; // Perform operations - 10 times on default collection, 10 times on col1 // collection for (int i = 0; i < 10; ++i) { std::string base_key = butil::string_printf("shared_thread_op_%d_thread_id_%d", i, shared_args->thread_id); // CRUD operations on default collection perform_crud_operations_default(*shared_couchbase_ops, base_key, shared_args->stats); // CRUD operations on col1 collection perform_crud_operations_col1(*shared_couchbase_ops, base_key, shared_args->stats); // Small delay between operations bthread_usleep(10000); // 10ms } return NULL; } // Print simple statistics void print_stats() { g_stats.aggregate_stats(); std::cout << std::endl; std::cout << CYAN << "=== TEST RESULTS ===" << RESET << std::endl; int total_attempted = g_stats.total.operations_attempted.load(); int total_successful = g_stats.total.operations_successful.load(); int total_failed = g_stats.total.operations_failed.load(); double success_rate = total_attempted > 0 ? (double)total_successful / total_attempted * 100.0 : 0.0; std::cout << GREEN << "Overall Performance:" << RESET << std::endl; std::cout << " Total Operations: " << total_attempted << std::endl; std::cout << " Successful: " << total_successful << " (" << success_rate << "%)" << std::endl; std::cout << " Failed: " << total_failed << std::endl; std::cout << std::endl; // Per-thread breakdown std::cout << YELLOW << "Per-Thread Performance:" << RESET << std::endl; for (int i = 0; i < NUM_THREADS; ++i) { const auto& stats = g_stats.per_thread_stats[i]; int attempted = stats.operations_attempted.load(); int successful = stats.operations_successful.load(); int failed = stats.operations_failed.load(); std::cout << " Thread " << i << ": " << attempted << " ops, " << successful << " success, " << failed << " failed" << std::endl; } std::cout << std::endl; } int main(int argc, char* argv[]) { std::cout << GREEN << "Starting Simple Multithreaded Couchbase Client" << RESET << std::endl; std::cout << YELLOW << "20 threads: 5 per bucket (testing0, testing1, testing2, testing3)" << RESET << std::endl; std::cout << BLUE << "Each thread performs CRUD operations on default collection and " "col1 collection" << RESET << std::endl; // Create threads and arguments std::vector threads(NUM_THREADS); std::vector args(NUM_THREADS); // Assign threads to buckets for (int i = 0; i < NUM_THREADS; ++i) { args[i].thread_id = i; args[i].bucket_id = i / THREADS_PER_BUCKET; args[i].bucket_name = g_config.bucket_names[args[i].bucket_id]; args[i].stats = &g_stats.per_thread_stats[i]; } // Print thread assignments std::cout << "Thread Assignments:" << RESET << std::endl; for (int i = 0; i < NUM_THREADS; ++i) { std::cout << "Thread " << i << " -> Bucket: " << args[i].bucket_name << std::endl; } std::cout << std::endl; // Start all threads for (int i = 0; i < NUM_THREADS; ++i) { if (bthread_start_background(&threads[i], NULL, thread_worker, &args[i]) != 0) { std::cout << RED << "Failed to create thread " << i << RESET << std::endl; return -1; } } std::cout << GREEN << "All 20 threads started!" << RESET << std::endl; // Wait for all threads to complete std::cout << YELLOW << "Waiting for all threads to complete..." << RESET << std::endl; for (int i = 0; i < NUM_THREADS; ++i) { bthread_join(threads[i], NULL); } std::cout << GREEN << "All threads completed!" << RESET << std::endl; // create a shared CouchbaseOperations instance brpc::CouchbaseOperations shared_couchbase_ops; brpc::CouchbaseOperations::Result result; result = shared_couchbase_ops.authenticate( g_config.username, g_config.password, "127.0.0.1:11210", "t0"); if(result.success){ std::cout << GREEN << "Shared CouchbaseOperations instance authenticated successfully!" << RESET << std::endl; } else { std::cout << RED << "Shared CouchbaseOperations instance authentication failed: " << result.error_message << RESET << std::endl; return -1; } for (int i = 0; i < NUM_THREADS; ++i) { args[i].thread_id = i; args[i].couchbase_ops = &shared_couchbase_ops; args[i].bucket_id = 0; args[i].bucket_name = "t0"; // args[i].stats = &g_stats.per_thread_stats[i]; } for(int i = 0; i < NUM_THREADS; ++i){ if (bthread_start_background(&threads[i], NULL, shared_object_thread_worker, &args[i]) != 0) { std::cout << RED << "Failed to create shared object thread " << i << RESET << std::endl; return -1; } } for(int i = 0; i < NUM_THREADS; ++i){ bthread_join(threads[i], NULL); } std::cout << GREEN << "All shared object threads completed!" << RESET << std::endl; // Print statistics print_stats(); return 0; }