#include<condition_variable>#include<functional>#include<iostream>#include<mutex>#include<optional>#include<queue>#include<thread>#include<vector>// ============================================// Channel: The core of CSP model// A typed, thread-safe, bounded channel// Similar to Go's buffered channel// ============================================template<typenameT>classChannel{public:// ============================================// Constructor// capacity = 0 means unbuffered (synchronous)// capacity > 0 means buffered (async up to capacity)// ============================================explicitChannel(size_t capacity=0):capacity_(capacity),closed_(false){}// ============================================// Send a value into the channel// Blocks if buffer is full// Returns false if channel is closed// ============================================boolsend(T value){std::unique_lock<std::mutex>lock(mu_);// Wait until:// 1. There's room in buffer, OR// 2. Channel is closed// For unbuffered (capacity_=0): wait until someone is receivingcv_send_.wait(lock,[this](){returnclosed_||queue_.size()<capacity_||(capacity_==0&&receivers_waiting_>0);});// Can't send to closed channelif(closed_){returnfalse;}// Put value in queuequeue_.push(std::move(value));// Wake up one waiting receivercv_recv_.notify_one();// For unbuffered channel: wait until value is takenif(capacity_==0){cv_send_.wait(lock,[this](){returnqueue_.empty()||closed_;});}returntrue;}// ============================================// Receive a value from the channel// Blocks if buffer is empty// Returns std::nullopt if channel is closed and empty// ============================================std::optional<T>recv(){std::unique_lock<std::mutex>lock(mu_);// Increment waiting receivers (for unbuffered channel)++receivers_waiting_;cv_send_.notify_one();// Wake sender for unbuffered case// Wait until:// 1. There's data in buffer, OR// 2. Channel is closedcv_recv_.wait(lock,[this](){return!queue_.empty()||closed_;});--receivers_waiting_;// Channel closed and emptyif(queue_.empty()){returnstd::nullopt;}// Get value from queueT value=std::move(queue_.front());queue_.pop();// Wake up one waiting sendercv_send_.notify_one();returnvalue;}// ============================================// Close the channel// No more sends allowed, but can still receive remaining data// ============================================voidclose(){std::lock_guard<std::mutex>lock(mu_);closed_=true;cv_send_.notify_all();// Wake all senderscv_recv_.notify_all();// Wake all receivers}// ============================================// Check if channel is closed// ============================================boolis_closed()const{std::lock_guard<std::mutex>lock(mu_);returnclosed_;}private:mutablestd::mutex mu_;std::condition_variable cv_send_;// Senders wait on thisstd::condition_variable cv_recv_;// Receivers wait on thisstd::queue<T>queue_;// The buffersize_t capacity_;// Max buffer size (0 = unbuffered)boolclosed_;// Is channel closed?size_t receivers_waiting_=0;// Count of waiting receivers};// ============================================// Convenience operator for sending (like Go's ch <- value)// ============================================template<typenameT>Channel<T>&operator<<(Channel<T>&ch,T value){ch.send(std::move(value));returnch;}// ============================================// Convenience operator for receiving (like Go's value = <-ch)// ============================================template<typenameT>Channel<T>&operator>>(Channel<T>&ch,T&value){autoresult=ch.recv();if(result){value=std::move(*result);}returnch;}
intmain(){// Create a buffered channel with capacity 2Channel<int>ch(2);// Producer threadstd::threadproducer([&ch](){for(inti=1;i<=5;i++){std::cout<<"Sending: "<<i<<std::endl;ch.send(i);std::cout<<"Sent: "<<i<<std::endl;}ch.close();// No more data});// Consumer threadstd::threadconsumer([&ch](){while(true){autovalue=ch.recv();if(!value){std::cout<<"Channel closed"<<std::endl;break;}std::cout<<"Received: "<<*value<<std::endl;std::this_thread::sleep_for(std::chrono::milliseconds(100));}});producer.join();consumer.join();return0;}
intmain(){Channel<std::string>ch(5);// Multiple producersstd::vector<std::thread>producers;for(inti=0;i<3;i++){producers.emplace_back([&ch,i](){for(intj=0;j<3;j++){std::string msg="Producer"+std::to_string(i)+"-Msg"+std::to_string(j);ch.send(msg);}});}// Multiple consumersstd::vector<std::thread>consumers;for(inti=0;i<2;i++){consumers.emplace_back([&ch,i](){while(true){automsg=ch.recv();if(!msg)break;std::cout<<"Consumer"<<i<<" got: "<<*msg<<std::endl;}});}// Wait for producersfor(auto&t:producers)t.join();// Close channel after all producers donech.close();// Wait for consumersfor(auto&t:consumers)t.join();return0;}
