diff --git a/nmux.cpp b/nmux.cpp index 7a3ee3e..56f650b 100644 --- a/nmux.cpp +++ b/nmux.cpp @@ -169,22 +169,26 @@ int main(int argc, char* argv[]) // waiting on this condition. They will be woken up with pthread_cond_broadcast() if new // data arrives. pthread_cond_t* wait_condition = new pthread_cond_t; - if(!pthread_cond_init(wait_condition, NULL)) + if(pthread_cond_init(wait_condition, NULL)) print_exit(MSG_START "(main thread) pthread_cond_init failed"); //cond_attrs is ignored by Linux for(;;) { + fprintf(stderr, "mainfor: selecting..."); //Let's wait until there is any new data to read, or any new connection! select(highfd, &select_fds, NULL, NULL, NULL); + fprintf(stderr, "selected.\n"); //Is there a new client connection? if( (new_socket = accept(listen_socket, (struct sockaddr*)&addr_cli, &addr_cli_len)) != -1) { + fprintf(stderr, "mainfor: accepted.\n"); //Close all finished clients for(int i=0;istatus == CS_THREAD_FINISHED) { + fprintf(stderr, "mainfor: client removed: %d\n", i); client_erase(clients[i]); clients.erase(clients.begin()+i); } @@ -193,7 +197,7 @@ int main(int argc, char* argv[]) //We're the parent, let's create a new client and initialize it client_t* new_client = new client_t; new_client->error = 0; - memcpy(&new_client->addr, &addr_cli, sizeof(new_client->addr)); + memcpy(&new_client->addr, &addr_cli, sizeof(struct sockaddr_in)); new_client->socket = new_socket; new_client->status = CS_CREATED; new_client->tsmthread = pool->register_thread(); @@ -201,7 +205,7 @@ int main(int argc, char* argv[]) pthread_mutex_init(&new_client->wait_mutex, NULL); new_client->wait_condition = wait_condition; new_client->sleeping = 0; - if(pthread_create(&new_client->thread, NULL, client_thread , (void*)&new_client)<0) + if(pthread_create(&new_client->thread, NULL, client_thread , (void*)&new_client)==0) { clients.push_back(new_client); fprintf(stderr, MSG_START "(main thread) pthread_create() done, clients now: %d\n", clients.size()); @@ -209,6 +213,9 @@ int main(int argc, char* argv[]) else { fprintf(stderr, MSG_START "(main thread) pthread_create() failed.\n"); + pool->remove_thread(new_client->tsmthread); + pthread_mutex_destroy(&new_client->wait_mutex); + delete new_client; } } @@ -223,6 +230,7 @@ int main(int argc, char* argv[]) index_in_current_write_buffer = 0; } int retval = read(input_fd, current_write_buffer + index_in_current_write_buffer, bufsize - index_in_current_write_buffer); + fprintf(stderr, "mainfor: read %d\n", retval); if(retval>0) { index_in_current_write_buffer += retval; @@ -248,16 +256,22 @@ void client_erase(client_t* client) void* client_thread (void* param) { + fprintf(stderr, "client 0x%x: started!\n", (unsigned)param); client_t* this_client = (client_t*)param; this_client->status = CS_THREAD_RUNNING; int retval; tsmpool* lpool = this_client->lpool; + fprintf(stderr, "client 0x%x: make maxfd... ", (unsigned)param); fd_set client_select_fds; + fprintf(stderr, "1"); int client_highfd = 0; FD_ZERO(&client_select_fds); + fprintf(stderr, "2 %d <= %d ", this_client->socket, FD_SETSIZE); FD_SET(this_client->socket, &client_select_fds); + fprintf(stderr, "3"); maxfd(&client_highfd, this_client->socket); + fprintf(stderr, "done.\n"); //Set client->socket to non-blocking if(set_nonblocking(this_client->socket)) @@ -271,8 +285,9 @@ void* client_thread (void* param) //Wait until there is any data to send. // If I haven't sent all the data from my last buffer, don't wait. // (Wait for the server process to wake me up.) - while(!pool_read_buffer || client_buffer_index == lpool->size) + while(!pool_read_buffer || client_buffer_index >= lpool->size) { + fprintf(stderr, "client 0x%x: waiting\n", (unsigned)param); char* pool_read_buffer = (char*)lpool->get_read_buffer(this_client->tsmthread); pthread_mutex_lock(&this_client->wait_mutex); this_client->sleeping = 1; @@ -280,10 +295,14 @@ void* client_thread (void* param) } //Wait for the socket to be available for write. + fprintf(stderr, "client 0x%x: selecting...", (unsigned)param); select(client_highfd, NULL, &client_select_fds, NULL, NULL); + fprintf(stderr, "done.\n"); //Read data from global tsmpool and write it to client socket + fprintf(stderr, "client 0x%x: sending...", (unsigned)param); int ret = send(this_client->socket, pool_read_buffer + client_buffer_index, lpool->size - client_buffer_index, 0); + fprintf(stderr, "done.\n"); if(ret == -1) { switch(errno) diff --git a/nmux.h b/nmux.h index f12db48..8efaf5a 100644 --- a/nmux.h +++ b/nmux.h @@ -31,7 +31,7 @@ typedef struct client_s tsmthread_t* tsmthread; client_status_t status; //the following members are there to give access to some global variables inside the thread: - tsmpool* lpool; //local pool + tsmpool* lpool; int sleeping; pthread_cond_t* wait_condition; pthread_mutex_t wait_mutex; diff --git a/tsmpool.cpp b/tsmpool.cpp index a82fd7a..92d94cb 100644 --- a/tsmpool.cpp +++ b/tsmpool.cpp @@ -8,7 +8,7 @@ tsmpool::tsmpool(size_t size, int num) : this->ok = 1; this->lowest_read_index = -1; this->write_index = 0; - this->my_read_index = 0; + this->my_read_index = index_before(0); if (pthread_mutex_init(&this->mutex, NULL) != 0) { this->ok = 0; return; } for(int i=0; iok; } void* tsmpool::get_write_buffer() { //if(write_index==index_before(lowest_read_index)) return NULL; + pthread_mutex_lock(&this->mutex); void* to_return = buffers[write_index]; - write_index=index_next(write_index); + write_index = index_next(write_index); + pthread_mutex_unlock(&this->mutex); + return to_return; } tsmthread_t* tsmpool::register_thread() { if(!ok) return NULL; pthread_mutex_lock(&this->mutex); - tsmthread_t* thread = new tsmthread_t; + tsmthread_t* thread = new tsmthread_t(); thread->read_index = index_before(write_index); threads.push_back(thread); pthread_mutex_unlock(&this->mutex); @@ -53,8 +56,11 @@ int tsmpool::remove_thread(tsmthread_t* thread) void* tsmpool::get_read_buffer(tsmthread_t* thread) { + pthread_mutex_lock(&this->mutex); int* actual_read_index = (thread==NULL) ? &my_read_index : &thread->read_index; if(*actual_read_index==index_before(write_index)) return NULL; void* to_return = buffers[*actual_read_index]; *actual_read_index=index_next(*actual_read_index); + pthread_mutex_unlock(&this->mutex); + return to_return; } diff --git a/tsmpool.h b/tsmpool.h index b2ca44a..26f3197 100644 --- a/tsmpool.h +++ b/tsmpool.h @@ -35,6 +35,6 @@ public: tsmthread_t* register_thread(); int remove_thread(tsmthread_t* thread); void* get_read_buffer(tsmthread_t* thread); - int index_next(int index) { return (index+1==size)?0:index; } - int index_before(int index) { return (index-1<0)?size-1:index; } + int index_next(int index) { return (index+1==num)?0:index+1; } + int index_before(int index) { return (index-1<0)?num-1:index-1; } };