tcp_srv.cc

Source code for an Numeric Cmputer

		    *((u_long *) ((char *) temp_buffer + dpi_offset)) =
			htonl(is_last_writer);
		    dpi_offset += 4;
		    int is_last_reader =
			(dpi == diagreply->cdi->last_reader_dpi);
		    *((u_long *) ((char *) temp_buffer + dpi_offset)) =
			htonl(is_last_reader);
		    dpi_offset += 4;
		    dpi =
			(CMS_DIAG_PROC_INFO *) diagreply->cdi->dpis->
			get_next();
		}
	    }
	    *((u_long *) temp_buffer + 6) = htonl(dpi_count);
	    *((u_long *) temp_buffer + 7) = htonl(dpi_offset);
	    if (sendn
		(_client_tcp_port->socket_fd, temp_buffer, dpi_offset, 0,
		    dtimeout) < 0) {
		_client_tcp_port->errors++;
		return;
	    }
	}
	break;

    case REMOTE_CMS_GET_BUF_NAME_REQUEST_TYPE:
	{
	    REMOTE_GET_BUF_NAME_REQUEST namereq;
	    namereq.buffer_number = buffer_number;
	    REMOTE_GET_BUF_NAME_REPLY *namereply = NULL;
	    namereply =
		(REMOTE_GET_BUF_NAME_REPLY *) server->
		process_request(&namereq);
	    memset(temp_buffer, 0, 40);
	    if (NULL != namereply) {
		*((u_long *) temp_buffer) =
		    htonl(_client_tcp_port->serial_number);
		*((u_long *) temp_buffer + 1) = htonl(namereply->status);
		strncpy(temp_buffer + 8, namereply->name, 31);
		if (sendn
		    (_client_tcp_port->socket_fd, temp_buffer, 40, 0,
			dtimeout) < 0) {
		    _client_tcp_port->errors++;
		    return;
		}
	    } else {
		*((u_long *) temp_buffer) =
		    htonl(_client_tcp_port->serial_number);
		*((u_long *) temp_buffer + 1) =
		    htonl((unsigned long) CMS_SERVER_SIDE_ERROR);
		if (sendn
		    (_client_tcp_port->socket_fd, temp_buffer, 40, 0,
			dtimeout) < 0) {
		    _client_tcp_port->errors++;
		    return;
		}
	    }
	}
	break;

    case REMOTE_CMS_BLOCKING_READ_REQUEST_TYPE:
	{
	    TCPSVR_BLOCKING_READ_REQUEST *blocking_read_req;

#ifdef NO_THREADS
	    if (NULL == _client_tcp_port->blocking_read_req) {
		_client_tcp_port->blocking_read_req =
		    new TCPSVR_BLOCKING_READ_REQUEST();
	    }
	    blocking_read_req = _client_tcp_port->blocking_read_req;
#else
	    blocking_read_req;
	    = new TCPSVR_BLOCKING_READ_REQUEST();
#endif
	    blocking_read_req->buffer_number = buffer_number;
	    blocking_read_req->access_type =
		ntohl(*((u_long *) temp_buffer + 3));
	    blocking_read_req->last_id_read =
		ntohl(*((u_long *) temp_buffer + 4));
	    total_subdivisions = 1;
	    if (max_total_subdivisions > 1) {
		total_subdivisions =
		    server->get_total_subdivisions(buffer_number);
	    }
	    if (total_subdivisions > 1) {
		if (recvn
		    (_client_tcp_port->socket_fd,
			(char *) (((u_long *) temp_buffer) + 5), 8, 0, -1,
			NULL) < 0) {
		    rcs_print_error
			("Can not read from client port (%d) from %s\n",
			_client_tcp_port->socket_fd,
			inet_ntoa(_client_tcp_port->address.sin_addr));
		    _client_tcp_port->errors++;
		    return;
		}
		blocking_read_req->subdiv =
		    ntohl(*((u_long *) temp_buffer + 6));
	    } else {
		if (recvn
		    (_client_tcp_port->socket_fd,
			(char *) (((u_long *) temp_buffer) + 5), 4, 0, -1,
			NULL) < 0) {
		    rcs_print_error
			("Can not read from client port (%d) from %s\n",
			_client_tcp_port->socket_fd,
			inet_ntoa(_client_tcp_port->address.sin_addr));
		    _client_tcp_port->errors++;
		    return;
		}
	    }
	    blocking_read_req->timeout_millis =
		ntohl(*((u_long *) temp_buffer + 5));
	    blocking_read_req->server = server;
	    blocking_read_req->remport = this;
	    _client_tcp_port->blocking = 1;
	    blocking_read_req->_client_tcp_port = _client_tcp_port;
#ifdef POSIX_THREADS
	    int thr_retval = pthread_create(&(_client_tcp_port->threadId),	/* ptr to new-thread-id */
		NULL,		// pthread_attr_t *, ptr to attributes
		tcpsvr_handle_blocking_request,	// start_func
		blocking_read_req	// arg for start_func
		);
	    if (thr_retval != 0) {
		_client_tcp_port->blocking = 0;
		rcs_print_error("pthread_create error: thr_retval = %d\n",
		    thr_retval);
		rcs_print_error("pthread_create error: %d %s\n", errno,
		    strerror(errno));
		*((u_long *) temp_buffer) =
		    htonl(_client_tcp_port->serial_number);
		*((u_long *) temp_buffer + 1) =
		    htonl((unsigned long) CMS_SERVER_SIDE_ERROR);
		*((u_long *) temp_buffer + 2) = htonl(0);	/* size */
		*((u_long *) temp_buffer + 3) = htonl(0);	/* write_id */
		*((u_long *) temp_buffer + 4) = htonl(0);	/* was_read */
		sendn(_client_tcp_port->socket_fd, temp_buffer, 20, 0,
		    dtimeout);
		return;
	    }
#else
#ifdef NO_THREADS
	    int fork_ret = fork();
	    switch (fork_ret) {
	    case 0:		// child
		_client_tcp_port->threadId = getpid();
		tcpsvr_handle_blocking_request(blocking_read_req);
		exit(0);
		break;

	    case -1:		// Error
		rcs_print_error("fork error: %d %s\n", errno,
		    strerror(errno));
		_client_tcp_port->blocking = 0;
		*((u_long *) temp_buffer) =
		    htonl(_client_tcp_port->serial_number);
		*((u_long *) temp_buffer + 1) =
		    htonl((unsigned long) CMS_SERVER_SIDE_ERROR);
		*((u_long *) temp_buffer + 2) = htonl(0);	/* size */
		*((u_long *) temp_buffer + 3) = htonl(0);	/* write_id */
		*((u_long *) temp_buffer + 4) = htonl(0);	/* was_read */
		sendn(_client_tcp_port->socket_fd, temp_buffer, 20, 0,
		    dtimeout);
		break;

	    default:		// parent;
		_client_tcp_port->threadId = fork_ret;
		break;
	    }
#else
	    rcs_print_error
		("Blocking read not supported on this platform.\n");
	    *((u_long *) temp_buffer) =
		htonl(_client_tcp_port->serial_number);
	    *((u_long *) temp_buffer + 1) =
		htonl((unsigned long) CMS_SERVER_SIDE_ERROR);
	    *((u_long *) temp_buffer + 2) = htonl(0);	/* size */
	    *((u_long *) temp_buffer + 3) = htonl(0);	/* write_id */
	    *((u_long *) temp_buffer + 4) = htonl(0);	/* was_read */
	    sendn(_client_tcp_port->socket_fd, temp_buffer, 20, 0, dtimeout);
	    return;

#endif
#endif
	    tcpsvr_threads_created++;
	}
	break;

    case REMOTE_CMS_READ_REQUEST_TYPE:
	server->read_req.buffer_number = buffer_number;
	server->read_req.access_type = ntohl(*((u_long *) temp_buffer + 3));
	server->read_req.last_id_read = ntohl(*((u_long *) temp_buffer + 4));
	server->read_reply =
	    (REMOTE_READ_REPLY *) server->process_request(&server->read_req);
	if (max_total_subdivisions > 1) {
	    total_subdivisions =
		server->get_total_subdivisions(buffer_number);
	}
	if (total_subdivisions > 1) {
	    if (recvn
		(_client_tcp_port->socket_fd,
		    (char *) (((u_long *) temp_buffer) + 5), 4, 0, -1,
		    NULL) < 0) {
		rcs_print_error
		    ("Can not read from client port (%d) from %s\n",
		    _client_tcp_port->socket_fd,
		    inet_ntoa(_client_tcp_port->address.sin_addr));
		_client_tcp_port->errors++;
		return;
	    }
	    server->read_req.subdiv = ntohl(*((u_long *) temp_buffer + 5));
	} else {
	    server->read_req.subdiv = 0;
	}
	if (NULL == server->read_reply) {
	    rcs_print_error("Server could not process request.\n");
	    *((u_long *) temp_buffer) =
		htonl(_client_tcp_port->serial_number);
	    *((u_long *) temp_buffer + 1) =
		htonl((unsigned long) CMS_SERVER_SIDE_ERROR);
	    *((u_long *) temp_buffer + 2) = htonl(0);	/* size */
	    *((u_long *) temp_buffer + 3) = htonl(0);	/* write_id */
	    *((u_long *) temp_buffer + 4) = htonl(0);	/* was_read */
	    sendn(_client_tcp_port->socket_fd, temp_buffer, 20, 0, dtimeout);
	    return;
	}
	*((u_long *) temp_buffer) = htonl(_client_tcp_port->serial_number);
	*((u_long *) temp_buffer + 1) = htonl(server->read_reply->status);
	*((u_long *) temp_buffer + 2) = htonl(server->read_reply->size);
	*((u_long *) temp_buffer + 3) = htonl(server->read_reply->write_id);
	*((u_long *) temp_buffer + 4) = htonl(server->read_reply->was_read);
	if (server->read_reply->size < (0x2000 - 20)
	    && server->read_reply->size > 0) {
	    memcpy(temp_buffer + 20, server->read_reply->data,
		server->read_reply->size);
	    if (sendn
		(_client_tcp_port->socket_fd, temp_buffer,
		    20 + server->read_reply->size, 0, dtimeout) < 0) {
		_client_tcp_port->errors++;
		return;
	    }
	} else {
	    if (sendn
		(_client_tcp_port->socket_fd, temp_buffer, 20, 0,
		    dtimeout) < 0) {
		_client_tcp_port->errors++;
		return;
	    }
	    if (server->read_reply->size > 0) {
		if (sendn
		    (_client_tcp_port->socket_fd, server->read_reply->data,
			server->read_reply->size, 0, dtimeout) < 0) {
		    _client_tcp_port->errors++;
		    return;
		}
	    }
	}
	break;

    case REMOTE_CMS_WRITE_REQUEST_TYPE:
	server->write_req.buffer_number = buffer_number;
	server->write_req.access_type = ntohl(*((u_long *) temp_buffer + 3));
	server->write_req.size = ntohl(*((u_long *) temp_buffer + 4));
	total_subdivisions = 1;
	if (max_total_subdivisions > 1) {
	    total_subdivisions =
		server->get_total_subdivisions(buffer_number);
	}
	if (total_subdivisions > 1) {
	    if (recvn
		(_client_tcp_port->socket_fd,
		    (char *) (((u_long *) temp_buffer) + 5), 4, 0, -1,
		    NULL) < 0) {
		rcs_print_error
		    ("Can not read from client port (%d) from %s\n",
		    _client_tcp_port->socket_fd,
		    inet_ntoa(_client_tcp_port->address.sin_addr));
		_client_tcp_port->errors++;
		return;
	    }
	    server->write_req.subdiv = ntohl(*((u_long *) temp_buffer + 5));
	} else {
	    server->write_req.subdiv = 0;
	}
	if (server->write_req.size > 0) {
	    if (recvn
		(_client_tcp_port->socket_fd, server->write_req.data,
		    server->write_req.size, 0, -1, NULL) < 0) {
		_client_tcp_port->errors++;
		return;
	    }
	}
	server->write_reply =
	    (REMOTE_WRITE_REPLY *) server->process_request(&server->
	    write_req);
	if (min_compatible_version < 2.58 && min_compatible_version > 1e-6
	    || server->write_reply->confirm_write) {
	    if (NULL == server->write_reply) {
		rcs_print_error("Server could not process request.\n");
		*((u_long *) temp_buffer) =
		    htonl(_client_tcp_port->serial_number);
		*((u_long *) temp_buffer + 1) =
		    htonl((unsigned long) CMS_SERVER_SIDE_ERROR);
		*((u_long *) temp_buffer + 2) = htonl(0);	/* was_read */
		sendn(_client_tcp_port->socket_fd, temp_buffer, 12, 0,
		    dtimeout);
		return;
	    }
	    *((u_long *) temp_buffer) =
		htonl(_client_tcp_port->serial_number);
	    *((u_long *) temp_buffer + 1) =
		htonl(server->write_reply->status);
	    *((u_long *) temp_buffer + 2) =
		htonl(server->write_reply->was_read);
	    if (sendn
		(_client_tcp_port->socket_fd, temp_buffer, 12, 0,
		    dtimeout) < 0) {
		_client_tcp_port->errors++;
	    }
	} else {
	    if (NULL == server->write_reply) {
		rcs_print_error("Server could not process request.\n");
	    }
	}
	break;

    case REMOTE_CMS_CHECK_IF_READ_REQUEST_TYPE:
	server->check_if_read_req.buffer_number = buffer_number;
	server->check_if_read_req.subdiv =
	    ntohl(*((u_long *) temp_buffer + 3));
	server->check_if_read_reply =
	    (REMOTE_CHECK_IF_READ_REPLY *) server->process_request(&server->
	    check_if_read_req);
	if (NULL == server->check_if_read_reply) {
	    rcs_print_error("Server could not process request.\n");
	    *((u_long *) temp_buffer) =
		htonl(_client_tcp_port->serial_number);
	    *((u_long *) temp_buffer + 1) =
		htonl((unsigned long) CMS_SERVER_SIDE_ERROR);
	    *((u_long *) temp_buffer + 2) = htonl(0);	/* was_read */
	    sendn(_client_tcp_port->socket_fd, temp_buffer, 12, 0, dtimeout);
	    return;
	}
	*((u_long *) temp_buffer) = htonl(_client_tcp_port->serial_number);
	*((u_long *) temp_buffer + 1) =
	    htonl(server->check_if_read_reply->status);
	*((u_long *) temp_buffer + 2) =
	    htonl(server->check_if_read_reply->was_read);
	if (sendn(_client_tcp_port->socket_fd, temp_buffer, 12, 0, dtimeout) <
	    0) {
	    _client_tcp_port->errors++;
	}
	break;

    case REMOTE_CMS_GET_MSG_COUNT_REQUEST_TYPE:
	server->get_msg_count_req.buffer_number = buffer_number;
	server->get_msg_count_req.subdiv =
	    ntohl(*((u_long *) temp_buffer + 3));
	server->get_msg_count_reply =
	    (REMOTE_GET_MSG_COUNT_REPLY *) server->process_request(&server->
	    get_msg_count_req);
	if (NULL == server->get_msg_count_reply) {
	    rcs_print_error("Server could not process request.\n");
	    *((u_long *) temp_buffer) =
		htonl(_client_tcp_port->serial_number);
	    *((u_long *) temp_buffer + 1) =
		htonl((unsigned long) CMS_SERVER_SIDE_ERROR);
	    *((u_long *) temp_buffer + 2) = htonl(0);	/* was_read */
	    sendn(_client_tcp_port->socket_fd, temp_buffer, 12, 0, dtimeout);
	    return;
	}
	*((u_long *) temp_buffer) = htonl(_client_tcp_port->serial_number);
	*((u_long *) temp_buffer + 1) =
	    htonl(server->get_msg_count_reply->status);
	*((u_long *) temp_buffer + 2) =
	    htonl(server->get_msg_count_reply->count);
	if (sendn(_client_tcp_port->socket_fd, temp_buffer, 12, 0, dtimeout) <
	    0) {
	    _client_tcp_port->errors++;
	}
	break;

    case REMOTE_CMS_GET_QUEUE_LENGTH_REQUEST_TYPE:
	server->get_queue_length_req.buffer_number = buffer_number;
	server->get_queue_length_req.subdiv =
	    ntohl(*((u_long *) temp_buffer + 3));
	server->get_queue_length_reply =
	    (REMOTE_GET_QUEUE_LENGTH_REPLY *) server->
	    process_request(&server->get_queue_length_req);
	if (NULL == server->get_queue_length_reply) {
	    rcs_print_error("Server could not process request.\n");
	    *((u_long *) temp_buffer) =
		htonl(_client_tcp_port->serial_number);
	    *((u_long *) temp_buffer + 1) =
		htonl((unsigned long) CMS_SERVER_SIDE_ERROR);
	    *((u_long *) temp_buffer + 2) = htonl(0);	/* was_read */
	    sendn(_client_tcp_port->socket_fd, temp_buffer, 12, 0, dtimeout);
	    return;
	}
	*((u_long *) temp_buffer) = htonl(_client_tcp_port->serial_number);
	*((u_long *) temp_buffer + 1) =
	    htonl(server->get_queue_length_reply->status);
	*((u_long *) temp_buffer + 2) =
	    htonl(server->get_queue_length_reply->queue_length);
	if (sendn(_client_tcp_port->socket_fd, temp_buffer, 12, 0, dtimeout) <
	    0) {
	    _client_tcp_port->errors++;
	}
	break;

    case REMOTE_CMS_GET_SPACE_AVAILABLE_REQUEST_TYPE:
	server->get_space_available_req.buffer_number = buffer_number;
	server->get_space_available_req.subdiv =
	    ntohl(*((u_long *) temp_buffer + 3));
	server->get_space_available_reply =
	    (REMOTE_GET_SPACE_AVAILABLE_REPLY *) server->
	    process_request(&server->get_space_available_req);
	if (NULL == server->get_space_available_reply) {
	    rcs_print_error("Server could not process request.\n");
	    *((u_long *) temp_buffer) =
		htonl(_client_tcp_port->serial_number);
	    *((u_long *) temp_buffer + 1) =
		htonl((unsigned long) CMS_SERVER_SIDE_ERROR);
	    *((u_long *) temp_buffer + 2) = htonl(0);	/* was_read */
	    sendn(_client_tcp_port->socket_fd, temp_buffer, 12, 0, dtimeout);
	    return;
	}
	*((u_long *) temp_buffer) = htonl(_client_tcp_port->serial_number);
	*((u_long *) temp_buffer + 1) =
	    htonl(server->get_space_available_reply->status);
	*((u_long *) temp_buffer + 2) =
	    htonl(server->get_space_available_reply->space_available);
	if (sendn(_client_tcp_port->socket_fd, temp_buffer, 12, 0, dtimeout) <
	    0) {
	    _client_tcp_port->errors++;
	}
	break;

    case REMOTE_CMS_CLEAR_REQUEST_TYPE:
	server->clear_req.buffer_number = buffer_number;
	server->clear_req.subdiv = ntohl(*((u_long *) temp_buffer + 3));
	server->clear_reply =
	    (REMOTE_CLEAR_REPLY *) server->process_request(&server->
	    clear_req);
	if (NULL == server->clear_reply) {
	    rcs_print_error("Server could not process request.\n");
	    *((u_long *) temp_buffer) =
		htonl(_client_tcp_port->serial_number);
	    *((u_long *) temp_buffer + 1) =
		htonl((u_long) CMS_SERVER_SIDE_ERROR);
	    sendn(_client_tcp_port->socket_fd, temp_buffer, 8, 0, dtimeout);
	    return;
	}
	*((u_long *) temp_buffer) = htonl(_client_tcp_port->serial_number);
	*((u_long *) temp_buffer + 1) = htonl(server->clear_reply->status);
	if (sendn(_client_tcp_port->socket_fd, temp_buffer, 8, 0, dtimeout) <
	    0) {
	    _client_tcp_port->errors++;
	}
	break;

    case REMOTE_CMS_CLEAN_REQUEST_TYPE:
	server->spawner_pid = server->server_pid;