Switched from select() to poll(), connections now confirm
that incoming data matches SBD format.
This commit is contained in:
cdramey
296
src/wstationd.c
296
src/wstationd.c
@ -6,7 +6,6 @@
|
||||
#include <stdlib.h>
|
||||
#include <limits.h>
|
||||
#include <fcntl.h>
|
||||
#include <sys/select.h>
|
||||
#include <sys/types.h>
|
||||
#include <sys/socket.h>
|
||||
#include <sys/time.h>
|
||||
@ -14,13 +13,37 @@
|
||||
#include <arpa/inet.h>
|
||||
#include <libgen.h>
|
||||
#include <getopt.h>
|
||||
#include <poll.h>
|
||||
#include "wstationd.h"
|
||||
|
||||
#define BUFFERSIZE 4096
|
||||
|
||||
// As of the writing of this daemon, SBD maximum message length
|
||||
// is hardware limited to 1960 bytes, however, the SBD format
|
||||
// has a maximum size of 64k per burst. We add three bytes
|
||||
// for the SBD header itself. This value largely
|
||||
// dictates the memory cost per connection.
|
||||
#define BUFFERSIZE 65539
|
||||
|
||||
// Hard limit on number of concurrent connections
|
||||
#define SLOTLIMIT 128
|
||||
|
||||
// Timeout for each connection, in seconds
|
||||
#define TIMEOUT 30
|
||||
|
||||
|
||||
static char *exec_name;
|
||||
static char* exec_name;
|
||||
|
||||
struct connection
|
||||
{
|
||||
// Buffer to store the data stream before the dump to file
|
||||
unsigned char buff[BUFFERSIZE];
|
||||
ssize_t len; // How many bytes have been written to the buffer
|
||||
struct timeval tv; // Time connection was made
|
||||
struct in_addr addr; // Address connection was made from
|
||||
};
|
||||
|
||||
|
||||
static int write_connection(struct connection *);
|
||||
|
||||
static void print_help()
|
||||
{
|
||||
@ -32,7 +55,7 @@ static void print_help()
|
||||
}
|
||||
|
||||
|
||||
int main(int argc, char *argv[])
|
||||
int main(int argc, char* argv[])
|
||||
{
|
||||
const struct option longopts[] = {
|
||||
{ "port", required_argument, NULL, 'p' },
|
||||
@ -41,14 +64,20 @@ int main(int argc, char *argv[])
|
||||
{ "help", no_argument, NULL, 'h' },
|
||||
{ NULL, 0, NULL, 0 }
|
||||
};
|
||||
|
||||
|
||||
int flags = 0;
|
||||
int sockfd = 0;
|
||||
int port = 10800; // Default port
|
||||
uint32_t address = INADDR_ANY; // Default listen address
|
||||
|
||||
struct connection* conns[SLOTLIMIT];
|
||||
memset(conns, 0, sizeof(struct connection*) * SLOTLIMIT);
|
||||
struct pollfd fds[SLOTLIMIT+1];
|
||||
memset(&fds, 0, sizeof(struct pollfd) * (SLOTLIMIT + 1));
|
||||
|
||||
exec_name = basename(argv[0]);
|
||||
|
||||
for(int ch; (ch = getopt_long(argc, argv, "b:p:vh0", longopts, NULL)) != -1;){
|
||||
for(int ch; (ch = getopt_long(argc, argv, "l:b:p:vh0", longopts, NULL)) != -1;){
|
||||
switch(ch){
|
||||
case 'p':
|
||||
port = (int)strtol(optarg, NULL, 10);
|
||||
@ -92,6 +121,7 @@ int main(int argc, char *argv[])
|
||||
goto shutdown_error;
|
||||
}
|
||||
|
||||
|
||||
// Establish the socket
|
||||
sockfd = socket(AF_INET, SOCK_STREAM, 0);
|
||||
if(sockfd == -1){
|
||||
@ -103,8 +133,8 @@ int main(int argc, char *argv[])
|
||||
|
||||
// This sets REUSE on the socket so it's easily reallocated if
|
||||
// this program dies
|
||||
int sockflag = 1;
|
||||
setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &sockflag, sizeof(sockflag));
|
||||
flags = 1;
|
||||
setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &flags, sizeof(flags));
|
||||
|
||||
struct sockaddr_in sock;
|
||||
memset(&sock, 0, sizeof(sock));
|
||||
@ -121,133 +151,185 @@ int main(int argc, char *argv[])
|
||||
}
|
||||
|
||||
// Listen on port
|
||||
if(listen(sockfd, 50) != 0){
|
||||
if(listen(sockfd, 10) != 0){
|
||||
fprintf(stderr, "%s: cannot listen - %s\n",
|
||||
exec_name, strerror(errno)
|
||||
);
|
||||
goto shutdown_error;
|
||||
}
|
||||
|
||||
// Allocate space for the output fds
|
||||
int writefds[FD_SETSIZE];
|
||||
memset(&writefds, 0, sizeof(int) * FD_SETSIZE);
|
||||
|
||||
// Setup select variables
|
||||
fd_set readfds, masterfds;
|
||||
FD_ZERO(&masterfds);
|
||||
FD_SET(sockfd, &masterfds);
|
||||
// Add sockfd
|
||||
fds[0].fd = sockfd;
|
||||
fds[0].events = POLLIN;
|
||||
fds[0].revents = 0;
|
||||
|
||||
while(1){
|
||||
readfds = masterfds;
|
||||
if(select(FD_SETSIZE, &readfds, NULL, NULL, NULL) == -1){
|
||||
fprintf(stderr, "%s: socket select error - %s\n",
|
||||
int r = poll(fds, (SLOTLIMIT + 1), 1000);
|
||||
|
||||
if(r == -1){
|
||||
fprintf(stderr, "%s: socket poll error - %s\n",
|
||||
exec_name, strerror(errno)
|
||||
);
|
||||
goto shutdown_error;
|
||||
}
|
||||
|
||||
for(int i = 0; i < FD_SETSIZE; i++){
|
||||
if(FD_ISSET(i, &readfds)){
|
||||
// Deal with new connections
|
||||
if(i == sockfd){
|
||||
struct sockaddr_in client;
|
||||
socklen_t client_sz = sizeof(client);
|
||||
// Handle existing connections
|
||||
int slotfirst = 0, slotfree = 0;
|
||||
for(int i = 1; i <= SLOTLIMIT; i++){
|
||||
// Handle event
|
||||
if(fds[i].revents & POLLIN){
|
||||
ssize_t sz = recv(
|
||||
fds[i].fd, conns[i-1]->buff,
|
||||
(BUFFERSIZE - conns[i-1]->len), 0
|
||||
);
|
||||
|
||||
int connfd = accept(sockfd, (struct sockaddr*)&client, &client_sz);
|
||||
if(connfd == -1){
|
||||
fprintf(stderr, "%s: accept failed - %s\n",
|
||||
exec_name, strerror(errno)
|
||||
);
|
||||
continue;
|
||||
}
|
||||
if(sz > 0) conns[i-1]->len += sz;
|
||||
else {
|
||||
if(conns[i-1]->len > 3){
|
||||
uint16_t sbd_sz = *(uint16_t*)(&conns[i-1]->buff[1]);
|
||||
// Convert and add three bytes for the SBD header
|
||||
ssize_t expected = (ssize_t)htons(sbd_sz) + 3;
|
||||
|
||||
// Get current flags before setting nonblock
|
||||
int connflag;
|
||||
if((connflag = fcntl(connfd, F_GETFL)) == -1){
|
||||
fprintf(stderr, "%s: cannot get connection flags - %s\n",
|
||||
exec_name, strerror(errno)
|
||||
);
|
||||
close(connfd);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Set nonblocking on socket
|
||||
if(fcntl(connfd, F_SETFL, connflag | O_NONBLOCK) == -1){
|
||||
fprintf(stderr, "%s: cannot set connection flags - %s\n",
|
||||
exec_name, strerror(errno)
|
||||
);
|
||||
close(connfd);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Collect the time of the connection, process it into
|
||||
// a short string
|
||||
struct timeval tvtime;
|
||||
gettimeofday(&tvtime, NULL);
|
||||
struct tm* timeptr = localtime(&tvtime.tv_sec);
|
||||
char timebuff[20];
|
||||
strftime(&timebuff[0], 20, "%Y%m%d%H%M%S", timeptr);
|
||||
|
||||
// Take the short date/time connection string and add the client address
|
||||
// to create a filename
|
||||
char fnbuff[50];
|
||||
snprintf(&fnbuff[0], 50, "%s.%ld-%s.dat",
|
||||
timebuff, (long)tvtime.tv_usec,
|
||||
inet_ntoa(client.sin_addr)
|
||||
);
|
||||
|
||||
// Open a file to dump our data to
|
||||
if((writefds[connfd] = open(fnbuff, O_WRONLY | O_CREAT, 0644)) == -1){
|
||||
writefds[connfd] = 0;
|
||||
fprintf(stderr, "%s: cannot open - %s\n",
|
||||
exec_name, strerror(errno)
|
||||
);
|
||||
|
||||
close(connfd);
|
||||
} else {
|
||||
FD_SET(connfd, &masterfds);
|
||||
}
|
||||
|
||||
// Read from existing connections
|
||||
} else {
|
||||
// Do the read
|
||||
char connbuff[BUFFERSIZE];
|
||||
ssize_t connbuff_sz = recv(i, connbuff, BUFFERSIZE, 0);
|
||||
if(connbuff_sz > 0){
|
||||
// If the read fails, throw an error, and set the
|
||||
// buffer size to 0, which forces a disconnect.
|
||||
if(write(writefds[i], connbuff, connbuff_sz) == -1){
|
||||
fprintf(stderr, "%s: file write error - %s\n",
|
||||
exec_name, strerror(errno)
|
||||
);
|
||||
connbuff_sz = 0;
|
||||
// Iridium SBD format check. Is the protocol byte in place?
|
||||
// Is the reported size the same as the actual size written
|
||||
// in bytes? If so, write it out.
|
||||
if(expected == conns[i-1]->len && conns[i-1]->buff[0] == 1){
|
||||
write_connection(conns[i - 1]);
|
||||
}
|
||||
}
|
||||
|
||||
// If the buffer is empty (which with select(), will only happen
|
||||
// on disconnect, disconnect and cleanup.
|
||||
if(connbuff_sz == 0){
|
||||
close(writefds[i]);
|
||||
writefds[i] = 0;
|
||||
FD_CLR(i, &masterfds);
|
||||
close(i);
|
||||
} else if(connbuff_sz == -1){
|
||||
fprintf(stderr, "%s: recv error - %s\n",
|
||||
exec_name, strerror(errno)
|
||||
);
|
||||
FD_CLR(i, &masterfds);
|
||||
close(i);
|
||||
}
|
||||
free(conns[i - 1]);
|
||||
conns[i - 1] = NULL;
|
||||
|
||||
close(fds[i].fd);
|
||||
fds[i].fd = 0;
|
||||
fds[i].events = 0;
|
||||
}
|
||||
|
||||
// Sweep up dead connections
|
||||
} else if(fds[i].fd){
|
||||
struct timeval tv;
|
||||
gettimeofday(&tv, NULL);
|
||||
|
||||
if((tv.tv_sec - conns[i - 1]->tv.tv_sec) >= TIMEOUT){
|
||||
free(conns[i - 1]);
|
||||
conns[i - 1] = NULL;
|
||||
|
||||
close(fds[i].fd);
|
||||
fds[i].fd = fds[i].events = 0;
|
||||
}
|
||||
}
|
||||
|
||||
// Count total free slots
|
||||
if(!fds[i].fd){
|
||||
++slotfree;
|
||||
// Note the first available slot
|
||||
if(!slotfirst) slotfirst = i;
|
||||
}
|
||||
|
||||
// Clear events
|
||||
fds[i].revents = 0;
|
||||
}
|
||||
|
||||
// Handle new connections
|
||||
if((fds[0].revents & POLLIN) && slotfree){
|
||||
struct sockaddr_in client;
|
||||
socklen_t client_sz = sizeof(client);
|
||||
|
||||
int fd = accept(sockfd, (struct sockaddr*)&client, &client_sz);
|
||||
if(fd == -1){
|
||||
fprintf(stderr, "%s: accept failed - %s\n",
|
||||
exec_name, strerror(errno)
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Get current flags before setting nonblock
|
||||
if((flags = fcntl(fd, F_GETFL)) == -1){
|
||||
fprintf(stderr, "%s: cannot get connection flags - %s\n",
|
||||
exec_name, strerror(errno)
|
||||
);
|
||||
close(fd);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Set nonblocking on socket
|
||||
if(fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1){
|
||||
fprintf(stderr, "%s: cannot set connection flags - %s\n",
|
||||
exec_name, strerror(errno)
|
||||
);
|
||||
close(fd);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Allocate memory for the connection's state.
|
||||
conns[slotfirst - 1] = malloc(sizeof(struct connection));
|
||||
if(conns[slotfirst - 1] == NULL){
|
||||
fprintf(stderr, "%s: out of memory\n", exec_name);
|
||||
close(fd);
|
||||
continue;
|
||||
}
|
||||
|
||||
conns[slotfirst - 1]->len = 0; // reset buffer
|
||||
conns[slotfirst - 1]->addr = client.sin_addr; // save source ip
|
||||
gettimeofday(&conns[slotfirst - 1]->tv, NULL); // mark connection time
|
||||
|
||||
// Add to poll list
|
||||
fds[slotfirst].fd = fd;
|
||||
fds[slotfirst].events = POLLIN;
|
||||
fds[slotfirst].revents = 0;
|
||||
slotfree--;
|
||||
|
||||
fds[0].revents = 0;
|
||||
}
|
||||
|
||||
// Reset polling event on listener socket -
|
||||
// if there's no free slots, stop polling the listener
|
||||
if(slotfree) fds[0].events = POLLIN;
|
||||
else fds[0].events = 0;
|
||||
}
|
||||
|
||||
shutdown_clean:
|
||||
if(sockfd > 0) close(sockfd);
|
||||
for(int i = 0; i < SLOTLIMIT; i++){ if(conns[i] != NULL) free(conns[i]); }
|
||||
return 0;
|
||||
|
||||
shutdown_error:
|
||||
if(sockfd > 0) close(sockfd);
|
||||
return 1;
|
||||
}
|
||||
|
||||
|
||||
static int write_connection(struct connection *conn)
|
||||
{
|
||||
// Build a filename for the file we're writing based
|
||||
// on the time and the source IP address
|
||||
char timebuff[20];
|
||||
struct tm* timeptr = localtime(&conn->tv.tv_sec);
|
||||
strftime(&timebuff[0], 20, "%Y%m%d%H%M%S", timeptr);
|
||||
|
||||
char fnbuff[50];
|
||||
snprintf(&fnbuff[0], 50, "%s.%ld-%s.dat",
|
||||
timebuff, (long)conn->tv.tv_usec,
|
||||
inet_ntoa(conn->addr)
|
||||
);
|
||||
|
||||
int ofd;
|
||||
if((ofd = open(fnbuff, O_WRONLY | O_CREAT, 0644)) == -1){
|
||||
fprintf(stderr, "%s: cannot open - %s\n",
|
||||
exec_name, strerror(errno)
|
||||
);
|
||||
return 1;
|
||||
}
|
||||
|
||||
if(write(ofd, conn->buff, conn->len) == -1){
|
||||
fprintf(stderr, "%s: file write error - %s\n",
|
||||
exec_name, strerror(errno)
|
||||
);
|
||||
close(ofd);
|
||||
return 1;
|
||||
}
|
||||
|
||||
close(ofd);
|
||||
return 0;
|
||||
}
|
||||
|
||||
@ -1 +1 @@
|
||||
#define VERSION "0.1.0"
|
||||
#define VERSION "0.7.0"
|
||||
|
||||
Reference in New Issue
Block a user