I'm writing a pthread server which takes requests from clients and sends them back a bunch
of .ppm files. Everything seems to go well, but sometimes when I have just 1 client connected, when trying to read from
the file descriptor (for
the file), it says Bad file Descriptor. This doesn't make sense, since my int fd isn't -1, and
the file most certainly exists. Other times, I get this "Socket operation on nonsocket" error. This is weird because other times, it doesn't give me this error and everything works fine.
When trying to connect multiple clients, for some reason, it will only send correctly to one, and then
the other client gets
the bad file descriptor or "nonsocket" error, even though both threads are processing
the same messages and do
the same routines. Anyone have an idea why? Here's
the code that is giving me that error:
while(mqueue.head != mqueue.tail && count < dis_m){
printf("Sending to client %s: %s\n", pointer->id, pointer->message);
int fd;
fd = open(pointer->message, O_RDONLY);
char buf[58368];
int bytesRead;
printf("This is fd %d\n", fd);
bytesRead=read(fd,buf,58368);
send(pointer->socket,buf,bytesRead,0);
perror("Error:\n");
fflush(stdout);
close(fd);
mqueue.mcount--;
mqueue.head = mqueue.head->next;
free(pointer->message);
free(pointer);
pointer = mqueue.head;
count++;
}
printf("Sending %s\n", pointer->message);
int fd;
fd = open(pointer->message, O_RDONLY);
printf("This is fd %d\n", fd);
printf("I am hhere2\n");
char buf[58368];
int bytesRead;
bytesRead=read(fd,buf,58368);
send(pointer->socket,buf,bytesRead,0);
perror("Error:\n");
close(fd);
mqueue.mcount--;
if(mqueue.head != mqueue.tail){
mqueue.head = mqueue.head->next;
}
else{
mqueue.head->next = malloc(sizeof(struct message));
mqueue.head = mqueue.head->next;
mqueue.head->next = malloc(sizeof(struct message));
mqueue.tail = mqueue.head->next;
mqueue.head->message = NULL;
}
free(pointer->message);
free(pointer);
pthread_mutex_unlock(&numm);
pthread_mutex_unlock(&circ);
pthread_mutex_unlock(&slots);
The messages for both threads are
the same, being
of the form ./path/imageXX.ppm where XX is
the number that should go to
the client.
The file size
of each image is 58368 bytes.
Sometimes, this code hangs on
the read, and stops execution. I don't know this would be either, because
the file descriptor comes back as valid.
Thanks in advanced.
Edit:
Here's some sample output:
Sending to client a: ./support/images/sw90.ppm
This is fd 4
Error:
: Socket operation on non-socket
Sending to client a: ./support/images/sw91.ppm
This is fd 4
Error:
: Socket operation on non-socket
Sending ./support/images/sw92.ppm
This is fd 4
I am hhere2
Error:
: Socket operation on non-socket
My dispatcher has defeated evil
Sample with 2 clients (client b was serviced first)
Sending to client b: ./support/images/sw87.ppm
This is fd 6
Error:
: Success
Sending to client b: ./support/images/sw88.ppm
This is fd 6
Error:
: Success
Sending to client b: ./support/images/sw89.ppm
This is fd 6
Error:
: Success
This is fd 6
Error:
: Bad file descriptor
Sending to client a: ./support/images/sw85.ppm
This is fd 6
Error:
As you can see, who ever is serviced first in this instance can open
the files, but not
the 2nd person.
Edit2:
Full code. Sorry, its pretty long and terribly formatted.
#include <netinet/in.h>
#include <netinet/in.h>
#include <netdb.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <errno.h>
#include <stdio.h>
#include <unistd.h>
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include "ring.h"
/*
Version 1
Here is what is implemented so far:
The threads are created from
the arguments specified (number
of threads that is)
The server will lock and update variables based on how many clients are in
the system and such.
The socket that is opened when a new client connects, must be passed to
the threads. To do this, we need some sort
of global array. I did
this by specifying an int client and main_pool_busy, and two pointers poolsockets and nonpoolsockets.
My thinking on this was that when a new client enters
the system,
the server thread increments
the variable client. When a thread is finished with this client (after it sends it
the data),
the thread will decrement client and close
the socket. HTTP servers act this way sometimes (they terminate
the socket as soon as one transmission is sent). *Note down at bottom
After
the server portion increments
the client counter, we must open up a new socket (denoted by new_sd) and get this value to
the appropriate thread.
To do this, I created global array poolsockets, which will hold all
the socket descriptors for our pooled threads.
The server portion gets
the new socket descriptor, and places
the value in
the first spot
of the array that has a 0. We only place a value in this array IF:
1.
The variable main_pool_busy < worknum (If we have more clients in
the system than in our pool, it doesn't mean we should always create a new thread.
At
the end of this,
the server signals on
the condition variable clientin that a new client has arrived.
In our pooled thread, we then must walk this array and check
the array until we hit our first non-zero value. This is
the socket we will give to that thread.
The thread then changes
the array to have a zero here.
What if our all threads in our pool our busy?
If this is
the case, then we will know it because our threads in this pool will increment main_pool_busy by one when they are working on a request and decrement it when they are done. If main_pool_busy >= worknum, then we must dynamically create a new thread. Then, we must realloc
the size
of our
nonpoolsockets array by 1 int. We then add
the new socket descriptor to our pool.
Here's what we need to figure out:
NOTE* Each worker should generate 100 messages which specify
the worker thread ID, client socket descriptor and a copy
of the client message. Additionally, each message should include a message number, starting from 0 and incrementing for each subsequent message sent to
the same client.
I don't know how to keep track
of how many messages were to
the same client. Maybe we shouldn't close
the socket descriptor, but rather keep an array
of structs for each socket that includes how many messages they have been sent. Then,
the server adds
the struct,
the threads remove it, then
the threads add it back once they've serviced one request (unless
the count is 100).
-------------------------------------------------------------
CHANGES
Version 1
----------
NONE: this is
the first version.
*/
#define MAXSLOTS 30
#define dis_m 15 //problems with dis_m ==1
//Function prototypes
void inc_clients();
void init_mutex_stuff(pthread_t*, pthread_t*);
void *threadpool(void *);
void server(int);
void add_to_socket_pool(int);
void inc_busy();
void dec_busy();
void *dispatcher();
void create_message(long, int, int, char *, char *);
void init_ring();
void add_to_ring(char *, char *, int, int, int);
int socket_from_string(char *);
void add_to_head(char *);
void add_to_tail(char *);
struct message * reorder(struct message *, struct message *, int);
int get_threadid(char *);
void delete_socket_messages(int);
struct message * merge(struct message *, struct message *, int);
int get_request(char *, char *, char*);
/////////////////////
//Global mutexes and condition variables
pthread_mutex_t startservice;
pthread_mutex_t numclients;
pthread_mutex_t pool_sockets;
pthread_mutex_t nonpool_sockets;
pthread_mutex_t m_pool_busy;
pthread_mutex_t slots;
pthread_mutex_t numm;
pthread_mutex_t circ;
pthread_cond_t clientin;
pthread_cond_t m;
///////////////////////////////////////
//Global variables
int clients;
int main_pool_busy;
int * poolsockets, nonpoolsockets;
int worknum;
struct ring mqueue;
///////////////////////////////////////
int main(int argc, char ** argv){
//error handling if not enough arguments to program
if(argc != 3){
printf("Not enough arguments to server: ./server portnum NumThreadsinPool\n");
_exit(-1);
}
//Convert arguments from strings to integer values
int port = atoi(argv[1]);
worknum = atoi(argv[2]);
//Start server portion
server(port);
}
///////////////////////////////////////////////////////////////////////////////////////////////
//The listen server thread/////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////////////////////
void server(int port){
int sd, new_sd;
struct sockaddr_in name, cli_name;
int sock_opt_val = 1;
int cli_len;
pthread_t threads[worknum]; //create our pthread id array
pthread_t dis[1]; //create our dispatcher array (necessary to create thread)
init_mutex_stuff(threads, dis); //initialize mutexes and stuff
//Server setup ///////////////////////////////////////////////////////
if ((sd = socket (AF_INET, SOCK_STREAM, 0)) < 0) {
perror("(servConn): socket() error");
_exit (-1);
}
if (setsockopt (sd, SOL_SOCKET, SO_REUSEADDR, (char *) &sock_opt_val,
sizeof(sock_opt_val)) < 0) {
perror ("(servConn): Failed to set SO_REUSEADDR on INET socket");
_exit (-1);
}
name.sin_family = AF_INET;
name.sin_port = htons (port);
name.sin_addr.s_addr = htonl(INADDR_ANY);
if (bind (sd, (struct sockaddr *)&name, sizeof(name)) < 0) {
perror ("(servConn): bind() error");
_exit (-1);
}
listen (sd, 5);
//End
of server Setup //////////////////////////////////////////////////
for (;;) {
cli_len = sizeof (cli_name);
new_sd = accept (sd, (struct sockaddr *) &cli_name, &cli_len);
printf ("Assigning new socket descriptor: %d\n", new_sd);
inc_clients(); //New client has come in, increment clients
add_to_socket_pool(new_sd); //Add client to
the pool
of sockets
if (new_sd < 0) {
perror ("(servConn): accept() error");
_exit (-1);
}
}
pthread_exit(NULL); //Quit
}
//Adds
the new socket to
the array designated for pthreads in
the pool
void add_to_socket_pool(int socket){
pthread_mutex_lock(&m_pool_busy); //Lock so that we can check main_pool_busy
int i;
//If not all our main pool is busy, then allocate to one
of them
if(main_pool_busy < worknum){
pthread_mutex_unlock(&m_pool_busy); //unlock busy, we no longer need to hold it
pthread_mutex_lock(&pool_sockets); //Lock
the socket pool array so that we can edit it without worry
for(i = 0; i < worknum; i++){ //Find a poolsocket that is -1; then we should put
the real socket there. This value will be changed back to -1 when
the thread grabs
the sockfd
if(poolsockets[i] == -1){
poolsockets[i] = socket;
pthread_mutex_unlock(&pool_sockets); //unlock our pool array, we don't need it anymore
inc_busy(); //Incrememnt busy (locks
the mutex itself)
pthread_cond_signal(&clientin); //Signal first thread waiting on a client that a client needs to be serviced
break;
}
}
}
else{ //Dynamic thread creation goes here
pthread_mutex_unlock(&m_pool_busy);
}
}
//Increments
the client number. If client number goes over worknum, we must dynamically create new pthreads
void inc_clients(){
pthread_mutex_lock(&numclients);
clients++;
pthread_mutex_unlock(&numclients);
}
//Increments busy
void inc_busy(){
pthread_mutex_lock(&m_pool_busy);
main_pool_busy++;
pthread_mutex_unlock(&m_pool_busy);
}
//Initialize all
of our mutexes at
the beginning and create our pthreads
void init_mutex_stuff(pthread_t * threads, pthread_t * dis){
pthread_mutex_init(&startservice, NULL);
pthread_mutex_init(&numclients, NULL);
pthread_mutex_init(&pool_sockets, NULL);
pthread_mutex_init(&nonpool_sockets, NULL);
pthread_mutex_init(&m_pool_busy, NULL);
pthread_mutex_init(&circ, NULL);
pthread_cond_init (&clientin, NULL);
main_pool_busy = 0;
poolsockets = malloc(sizeof(int)*worknum);
int threadreturn; //error checking variables
long i = 0; //Loop and create pthreads
for(i; i < worknum; i++){
threadreturn = pthread_create(&threads[i], NULL, threadpool, (void *) i);
poolsockets[i] = -1;
if(threadreturn){
perror("Thread pool created unsuccessfully");
_exit(-1);
}
}
pthread_create(&dis[0], NULL, dispatcher, NULL);
}
//////////////////////////////////////////////////////////////////////////////////////////
/////////Main pool routines
/////////////////////////////////////////////////////////////////////////////////////////
void dec_busy(){
pthread_mutex_lock(&m_pool_busy);
main_pool_busy--;
pthread_mutex_unlock(&m_pool_busy);
}
void dec_clients(){
pthread_mutex_lock(&numclients);
clients--;
pthread_mutex_unlock(&numclients);
}
//This is what our threadpool pthreads will be running.
void *threadpool(void * threadid){
long id = (long) threadid; //Id
of this thread
int i;
int socket;
int counter = 0;
//Try and gain access to
the next client that comes in and wait until server signals that a client as arrived
while(1){
pthread_mutex_lock(&startservice); //lock start service (required for cond wait)
pthread_cond_wait(&clientin, &startservice); //wait for signal from server that client exists
pthread_mutex_unlock(&startservice); //unlock mutex.
pthread_mutex_lock(&pool_sockets); //Lock
the pool socket so we can get
the socket fd unhindered/interrupted
for(i = 0; i < worknum; i++){
if(poolsockets[i] != -1){
socket = poolsockets[i];
poolsockets[i] = -1;
pthread_mutex_unlock(&pool_sockets);
}
}
printf("Thread #%d is past getting
the socket\n", id);
int incoming = 1;
while(counter < 100 && incoming != 0){
char buffer[512];
bzero(buffer,512);
int startcounter = 0;
incoming = read(socket, buffer, 512);
if(buffer[0] != 0){
//client ID:priority:request:arguments
char id[100];
long prior;
char request[100];
char arg1[100];
char message[100];
char arg2[100];
char * point;
point = strtok(buffer, ":");
strcpy(id, point);
point = strtok(NULL, ":");
prior = atoi(point);
point = strtok(NULL, ":");
strcpy(request, point);
point = strtok(NULL, ":");
strcpy(arg1, point);
point = strtok(NULL, ":");
if(point != NULL){
strcpy(arg2, point);
}
int fd;
if(strcmp(request, "start_movie") == 0){
int count = 1;
while(count <= 100){
char temp[10];
snprintf(temp, 50, "%d\0", count);
strcpy(message, "./support/images/");
strcat(message, arg1);
strcat(message, temp);
strcat(message, ".ppm");
printf("This is message %s to %s\n", message, id);
count++;
add_to_ring(message, id, prior, counter, socket); //Adds our created message to
the ring
counter++;
}
printf("I'm out
of the loop\n");
}
else if(strcmp(request, "seek_movie") == 0){
int count = atoi(arg2);
while(count <= 100){
char temp[10];
snprintf(temp, 10, "%d\0", count);
strcpy(message, "./support/images/");
strcat(message, arg1);
strcat(message, temp);
strcat(message, ".ppm");
printf("This is message %s\n", message);
count++;
}
}
//create_message(id, socket, counter, buffer, message); //Creates our message from
the input from
the client. Stores it in buffer
}
else{
delete_socket_messages(socket);
break;
}
}
counter = 0;
close(socket);//Zero out counter again
}
dec_clients(); //client serviced, decrement clients
dec_busy(); //thread finished, decrement busy
}
//Creates a message
void create_message(long threadid, int socket, int counter, char * buffer, char * message){
snprintf(message, strlen(buffer)+15, "%d:%d:%d:%s", threadid, socket, counter, buffer);
}
//Gets
the socket from
the message string (maybe I should just pass in
the socket to another method)
int socket_from_string(char * message){
char * substr1 = strstr(message, ":");
char * substr2 = substr1;
substr2++;
int occurance = strcspn(substr2, ":");
char sock[10];
strncpy(sock, substr2, occurance);
return atoi(sock);
}
//Adds message to our ring buffer's head
void add_to_head(char * message){
printf("Adding to head
of ring\n");
mqueue.head->message = malloc(strlen(message)+1); //Allocate space for message
strcpy(mqueue.head->message, message); //copy bytes into allocated space
}
//Adds our message to our ring buffer's tail
void add_to_tail(char * message){
printf("Adding to tail
of ring\n");
mqueue.tail->message = malloc(strlen(message)+1); //allocate space for message
strcpy(mqueue.tail->message, message); //copy bytes into allocated space
mqueue.tail->next = malloc(sizeof(struct message)); //allocate space for
the next message struct
}
//Adds a message to our ring
void add_to_ring(char * message, char * id, int prior, int mnum, int socket){
//printf("This is message %s:" , message);
pthread_mutex_lock(&circ); //Lock
the ring buffer
pthread_mutex_lock(&numm); //Lock
the message count (will need this to make sure we can't fill
the buffer over
the max slots)
if(mqueue.head->message == NULL){
add_to_head(message); //Adds it to head
mqueue.head->socket = socket; //Set message socket
mqueue.head->priority = prior; //Set its priority (thread id)
mqueue.head->mnum = mnum; //Set its message number (used for sorting)
mqueue.head->id = malloc(sizeof(id));
strcpy(mqueue.head->id, id);
}
else if(mqueue.tail->message == NULL){ //This is
the problem for dis_m 1 I'm pretty sure
add_to_tail(message);
mqueue.tail->socket = socket;
mqueue.tail->priority = prior;
mqueue.tail->mnum = mnum;
mqueue.tail->id = malloc(sizeof(id));
strcpy(mqueue.tail->id, id);
}
else{
mqueue.tail->next = malloc(sizeof(struct message));
mqueue.tail = mqueue.tail->next;
add_to_tail(message);
mqueue.tail->socket = socket;
mqueue.tail->priority = prior;
mqueue.tail->mnum = mnum;
mqueue.tail->id = malloc(sizeof(id));
strcpy(mqueue.tail->id, id);
}
mqueue.mcount++;
pthread_mutex_unlock(&circ);
if(mqueue.mcount >= dis_m){
pthread_mutex_unlock(&numm);
pthread_cond_signal(&m);
}
else{
pthread_mutex_unlock(&numm);
}
printf("out
of add to ring\n");
fflush(stdout);
}
//////////////////////////////////
//Dispatcher routines
/////////////////////////////////
void *dispatcher(){
init_ring();
while(1){
pthread_mutex_lock(&slots);
pthread_cond_wait(&m, &slots);
pthread_mutex_lock(&numm);
pthread_mutex_lock(&circ);
printf("Dispatcher to
the rescue!\n");
mqueue.head = reorder(mqueue.head, mqueue.tail, mqueue.mcount);
//printf("This is
the head %s\n", mqueue.head->message);
//printf("This is
the tail %s\n", mqueue.head->message);
fflush(stdout);
struct message * pointer = mqueue.head;
int count = 0;
while(mqueue.head != mqueue.tail && count < dis_m){
printf("Sending to client %s: %s\n", pointer->id, pointer->message);
int fd;
fd = open(pointer->message, O_RDONLY);
char buf[58368];
int bytesRead;
printf("This is fd %d\n", fd);
bytesRead=read(fd,buf,58368);
send(pointer->socket,buf,bytesRead,0);
perror("Error:\n");
fflush(stdout);
close(fd);
mqueue.mcount--;
mqueue.head = mqueue.head->next;
free(pointer->message);
free(pointer);
pointer = mqueue.head;
count++;
}
printf("Sending %s\n", pointer->message);
int fd;
fd = open(pointer->message, O_RDONLY);
printf("This is fd %d\n", fd);
printf("I am hhere2\n");
char buf[58368];
int bytesRead;
bytesRead=read(fd,buf,58368);
send(pointer->socket,buf,bytesRead,0);
perror("Error:\n");
close(fd);
mqueue.mcount--;
if(mqueue.head != mqueue.tail){
mqueue.head = mqueue.head->next;
}
else{
mqueue.head->next = malloc(sizeof(struct message));
mqueue.head = mqueue.head->next;
mqueue.head->next = malloc(sizeof(struct message));
mqueue.tail = mqueue.head->next;
mqueue.head->message = NULL;
}
free(pointer->message);
free(pointer);
pthread_mutex_unlock(&numm);
pthread_mutex_unlock(&circ);
pthread_mutex_unlock(&slots);
printf("My dispatcher has defeated evil\n");
}
}
void init_ring(){
mqueue.head = malloc(sizeof(struct message));
mqueue.head->next = malloc(sizeof(struct message));
mqueue.tail = mqueue.head->next;
mqueue.mcount = 0;
}
struct message * reorder(struct message * begin, struct message *
end, int num){
//printf("I am reordering for size %d\n", num);
fflush(stdout);
int i;
if(num == 1){
//printf("Begin: %s\n", begin->message);
begin->next = NULL;
return begin;
}
else{
struct message * left = begin;
struct message * right;
int middle = num/2;
for(i = 1; i < middle; i++){
left = left->next;
}
right = left -> next;
left -> next = NULL;
//printf("Begin: %s\nLeft: %s\nright: %s\nend:%s\n", begin->message, left->message, right->message, end->message);
left = reorder(begin, left, middle);
if(num%2 != 0){
right = reorder(right,
end, middle+1);
}
else{
right = reorder(right,
end, middle);
}
return merge(left, right, num);
}
}
struct message * merge(struct message * left, struct message * right, int num){
//printf("I am merginging! left: %s %d, right: %s %dnum: %d\n", left->message,left->priority, right->message, right->priority, num);
struct message * start, * point;
int lenL= 0;
int lenR = 0;
int flagL = 0;
int flagR = 0;
int count = 0;
int middle1 = num/2;
int middle2;
if(num%2 != 0){
middle2 = middle1+1;
}
else{
middle2 = middle1;
}
while(lenL < middle1 && lenR < middle2){
count++;
//printf("In here for count %d\n", count);
if(lenL == 0 && lenR == 0){
if(left->priority < right->priority){
start = left; //Set
the start point
point = left; //set our enum;
left = left->next; //move
the left pointer
point->next = NULL; //Set
the next node to NULL
lenL++;
}
else if(left->priority > right->priority){
start = right;
point = right;
right = right->next;
point->next = NULL;
lenR++;
}
else{
if(left->mnum < right->mnum){
////printf("This is where we are\n");
start = left; //Set
the start point
point = left; //set our enum;
left = left->next; //move
the left pointer
point->next = NULL; //Set
the next node to NULL
lenL++;
}
else{
start = right;
point = right;
right = right->next;
point->next = NULL;
lenR++;
}
}
}
else{
if(left->priority < right->priority){
point->next = left;
left = left->next; //move
the left pointer
point = point->next;
point->next = NULL; //Set
the next node to NULL
lenL++;
}
else if(left->priority > right->priority){
point->next = right;
right = right->next;
point = point->next;
point->next = NULL;
lenR++;
}
else{
if(left->mnum < right->mnum){
point->next = left; //set our enum;
left = left->next;
point = point->next;//move
the left pointer
point->next = NULL; //Set
the next node to NULL
lenL++;
}
else{
point->next = right;
right = right->next;
point = point->next;
point->next = NULL;
lenR++;
}
}
}
if(lenL == middle1){
flagL = 1;
break;
}
if(lenR == middle2){
flagR = 1;
break;
}
}
if(flagL == 1){
point->next = right;
point = point->next;
for(lenR; lenR< middle2-1; lenR++){
point = point->next;
}
point->next = NULL;
mqueue.tail = point;
}
else{
point->next = left;
point = point->next;
for(lenL; lenL< middle1-1; lenL++){
point = point->next;
}
point->next = NULL;
mqueue.tail = point;
}
//printf("This is
the start %s\n", start->message);
//printf("This is mqueue.tail %s\n", mqueue.tail->message);
return start;
}
void delete_socket_messages(int a){
}
