AFG/dump1090
3
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

Restructuring of network output side.

Browse source 
Mostly refactoring the common code that was duplicated
between the different output types so that there aren't
many copies floating around.

This introduces "struct net_writer" to store the state of a
particular type of output service - buffers, time of last write,
connection count etc. prepareWrite() / completeWrite() give access
to the buffer and handle the actual writes and flushing when needed.

Heartbeat and time-based flushing move into a generic periodic-work
function.

Update the SBS output code to use the new infrastructure. This makes
a big different to CPU use when under load.
This commit is contained in:
Oliver Jowett 2014-10-03 22:55:21 +01:00
parent 8d4f1a396c
commit 1769ac9006
4 changed files with 216 additions and 160 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
dump1090.c
View file

@ -70,7 +70,7 @@ void modesInitConfig(void) {
Modes.freq = MODES_DEFAULT_FREQ; Modes.freq = MODES_DEFAULT_FREQ;
Modes.ppm_error = MODES_DEFAULT_PPM; Modes.ppm_error = MODES_DEFAULT_PPM;
Modes.check_crc = 1; Modes.check_crc = 1;
Modes.net_heartbeat_rate = MODES_NET_HEARTBEAT_RATE; Modes.net_heartbeat_interval = MODES_NET_HEARTBEAT_INTERVAL;
Modes.net_output_sbs_port = MODES_NET_OUTPUT_SBS_PORT; Modes.net_output_sbs_port = MODES_NET_OUTPUT_SBS_PORT;
Modes.net_output_raw_port = MODES_NET_OUTPUT_RAW_PORT; Modes.net_output_raw_port = MODES_NET_OUTPUT_RAW_PORT;
Modes.net_input_raw_port = MODES_NET_INPUT_RAW_PORT; Modes.net_input_raw_port = MODES_NET_INPUT_RAW_PORT;
@ -97,9 +97,7 @@ void modesInit(void) {
if ( ((Modes.icao_cache = (uint32_t *) malloc(sizeof(uint32_t) * MODES_ICAO_CACHE_LEN * 2) ) == NULL) || if ( ((Modes.icao_cache = (uint32_t *) malloc(sizeof(uint32_t) * MODES_ICAO_CACHE_LEN * 2) ) == NULL) ||
((Modes.pFileData = (uint16_t *) malloc(MODES_ASYNC_BUF_SIZE) ) == NULL) || ((Modes.pFileData = (uint16_t *) malloc(MODES_ASYNC_BUF_SIZE) ) == NULL) ||
((Modes.magnitude = (uint16_t *) malloc(MODES_ASYNC_BUF_SIZE+MODES_PREAMBLE_SIZE+MODES_LONG_MSG_SIZE) ) == NULL) || ((Modes.magnitude = (uint16_t *) malloc(MODES_ASYNC_BUF_SIZE+MODES_PREAMBLE_SIZE+MODES_LONG_MSG_SIZE) ) == NULL) ||
((Modes.maglut = (uint16_t *) malloc(sizeof(uint16_t) * 256 * 256) ) == NULL) || ((Modes.maglut = (uint16_t *) malloc(sizeof(uint16_t) * 256 * 256) ) == NULL) )
((Modes.beastOut = (char *) malloc(MODES_RAWOUT_BUF_SIZE) ) == NULL) ||
((Modes.rawOut = (char *) malloc(MODES_RAWOUT_BUF_SIZE) ) == NULL) )
{ {
fprintf(stderr, "Out of memory allocating data buffer.\n"); fprintf(stderr, "Out of memory allocating data buffer.\n");
exit(1); exit(1);
@ -130,10 +128,10 @@ void modesInit(void) {
} }
// Limit the maximum requested raw output size to less than one Ethernet Block // Limit the maximum requested raw output size to less than one Ethernet Block
if (Modes.net_output_raw_size > (MODES_RAWOUT_BUF_FLUSH)) if (Modes.net_output_flush_size > (MODES_OUT_FLUSH_SIZE))
{Modes.net_output_raw_size = MODES_RAWOUT_BUF_FLUSH;} {Modes.net_output_flush_size = MODES_OUT_FLUSH_SIZE;}
if (Modes.net_output_raw_rate > (MODES_RAWOUT_BUF_RATE)) if (Modes.net_output_flush_interval > (MODES_OUT_FLUSH_INTERVAL))
{Modes.net_output_raw_rate = MODES_RAWOUT_BUF_RATE;} {Modes.net_output_flush_interval = MODES_OUT_FLUSH_INTERVAL;}
if (Modes.net_sndbuf_size > (MODES_NET_SNDBUF_MAX)) if (Modes.net_sndbuf_size > (MODES_NET_SNDBUF_MAX))
{Modes.net_sndbuf_size = MODES_NET_SNDBUF_MAX;} {Modes.net_sndbuf_size = MODES_NET_SNDBUF_MAX;}
@ -407,7 +405,6 @@ void showHelp(void) {
"--raw Show only messages hex values\n" "--raw Show only messages hex values\n"
"--net Enable networking\n" "--net Enable networking\n"
"--modeac Enable decoding of SSR Modes 3/A & 3/C\n" "--modeac Enable decoding of SSR Modes 3/A & 3/C\n"
"--net-beast TCP raw output in Beast binary format\n"
"--net-only Enable just networking, no RTL device or file used\n" "--net-only Enable just networking, no RTL device or file used\n"
"--net-http-port <port> HTTP server port (default: 8080)\n" "--net-http-port <port> HTTP server port (default: 8080)\n"
"--net-ri-port <port> TCP raw input listen port (default: 30001)\n" "--net-ri-port <port> TCP raw input listen port (default: 30001)\n"
@ -415,8 +412,8 @@ void showHelp(void) {
"--net-sbs-port <port> TCP BaseStation output listen port (default: 30003)\n" "--net-sbs-port <port> TCP BaseStation output listen port (default: 30003)\n"
"--net-bi-port <port> TCP Beast input listen port (default: 30004)\n" "--net-bi-port <port> TCP Beast input listen port (default: 30004)\n"
"--net-bo-port <port> TCP Beast output listen port (default: 30005)\n" "--net-bo-port <port> TCP Beast output listen port (default: 30005)\n"
"--net-ro-size <size> TCP raw output minimum size (default: 0)\n" "--net-ro-size <size> TCP output minimum size (default: 0)\n"
"--net-ro-rate <rate> TCP raw output memory flush rate (default: 0)\n" "--net-ro-interval <rate> TCP output memory flush rate in seconds (default: 0)\n"
"--net-heartbeat <rate> TCP heartbeat rate in seconds (default: 60 sec; 0 to disable)\n" "--net-heartbeat <rate> TCP heartbeat rate in seconds (default: 60 sec; 0 to disable)\n"
"--net-buffer <n> TCP buffer size 64Kb * (2^n) (default: n=0, 64Kb)\n" "--net-buffer <n> TCP buffer size 64Kb * (2^n) (default: n=0, 64Kb)\n"
"--lat <latitude> Reference/receiver latitude for surface posn (opt)\n" "--lat <latitude> Reference/receiver latitude for surface posn (opt)\n"
@ -572,8 +569,7 @@ void backgroundTasks(void) {
static time_t next_stats; static time_t next_stats;
if (Modes.net) { if (Modes.net) {
modesReadFromClients(); modesNetPeriodicWork();
modesNetCleanup();
} }
// If Modes.aircrafts is not NULL, remove any stale aircraft // If Modes.aircrafts is not NULL, remove any stale aircraft
@ -701,11 +697,13 @@ int main(int argc, char **argv) {
Modes.net = 1; Modes.net = 1;
Modes.net_only = 1; Modes.net_only = 1;
} else if (!strcmp(argv[j],"--net-heartbeat") && more) { } else if (!strcmp(argv[j],"--net-heartbeat") && more) {
Modes.net_heartbeat_rate = atoi(argv[++j]) * 15; Modes.net_heartbeat_interval = atoi(argv[++j]);
} else if (!strcmp(argv[j],"--net-ro-size") && more) { } else if (!strcmp(argv[j],"--net-ro-size") && more) {
Modes.net_output_raw_size = atoi(argv[++j]); Modes.net_output_flush_size = atoi(argv[++j]);
} else if (!strcmp(argv[j],"--net-ro-rate") && more) { } else if (!strcmp(argv[j],"--net-ro-rate") && more) {
Modes.net_output_raw_rate = atoi(argv[++j]); Modes.net_output_flush_interval = atoi(argv[++j]) / 15; // backwards compatibility
} else if (!strcmp(argv[j],"--net-ro-interval") && more) {
Modes.net_output_flush_interval = atoi(argv[++j]);
} else if (!strcmp(argv[j],"--net-ro-port") && more) { } else if (!strcmp(argv[j],"--net-ro-port") && more) {
if (Modes.beast) // Required for legacy backward compatibility if (Modes.beast) // Required for legacy backward compatibility
{Modes.net_output_beast_port = atoi(argv[++j]);;} {Modes.net_output_beast_port = atoi(argv[++j]);;}

44
dump1090.h
View file

@ -116,9 +116,9 @@
#define MODES_LONG_MSG_SIZE (MODES_LONG_MSG_SAMPLES * sizeof(uint16_t)) #define MODES_LONG_MSG_SIZE (MODES_LONG_MSG_SAMPLES * sizeof(uint16_t))
#define MODES_SHORT_MSG_SIZE (MODES_SHORT_MSG_SAMPLES * sizeof(uint16_t)) #define MODES_SHORT_MSG_SIZE (MODES_SHORT_MSG_SAMPLES * sizeof(uint16_t))
#define MODES_RAWOUT_BUF_SIZE (1500) #define MODES_OUT_BUF_SIZE (1500)
#define MODES_RAWOUT_BUF_FLUSH (MODES_RAWOUT_BUF_SIZE - 200) #define MODES_OUT_FLUSH_SIZE (MODES_OUT_BUF_SIZE - 256)
#define MODES_RAWOUT_BUF_RATE (1000) // 1000 * 64mS = 1 Min approx #define MODES_OUT_FLUSH_INTERVAL (60)
#define MODES_ICAO_CACHE_LEN 1024 // Power of two required #define MODES_ICAO_CACHE_LEN 1024 // Power of two required
#define MODES_ICAO_CACHE_TTL 60 // Time to live of cached addresses #define MODES_ICAO_CACHE_TTL 60 // Time to live of cached addresses
@ -165,7 +165,7 @@
#define MODES_INTERACTIVE_DELETE_TTL 300 // Delete from the list after 300 seconds #define MODES_INTERACTIVE_DELETE_TTL 300 // Delete from the list after 300 seconds
#define MODES_INTERACTIVE_DISPLAY_TTL 60 // Delete from display after 60 seconds #define MODES_INTERACTIVE_DISPLAY_TTL 60 // Delete from display after 60 seconds
#define MODES_NET_HEARTBEAT_RATE 900 // Each block is approx 65mS - default is > 1 min #define MODES_NET_HEARTBEAT_INTERVAL 60 // seconds
#define MODES_NET_SERVICES_NUM 6 #define MODES_NET_SERVICES_NUM 6
#define MODES_NET_INPUT_RAW_PORT 30001 #define MODES_NET_INPUT_RAW_PORT 30001
@ -237,6 +237,15 @@ struct stDF {
unsigned char msg[MODES_LONG_MSG_BYTES]; // the binary unsigned char msg[MODES_LONG_MSG_BYTES]; // the binary
} tDF; } tDF;
// Common writer state for all output sockets of one type
struct net_writer {
int socket; // listening socket FD, used to identify the owning service
int connections; // number of active clients
void *data; // shared write buffer, sized MODES_OUT_BUF_SIZE
int dataUsed; // number of bytes of write buffer currently used
time_t lastWrite; // time of last write to clients
};
// Program global state // Program global state
struct { // Internal state struct { // Internal state
pthread_t reader_thread; pthread_t reader_thread;
@ -270,16 +279,14 @@ struct { // Internal state
// Networking // Networking
char aneterr[ANET_ERR_LEN]; char aneterr[ANET_ERR_LEN];
struct client *clients; // Our clients struct client *clients; // Our clients
int sbsos; // SBS output listening socket
int ros; // Raw output listening socket
int ris; // Raw input listening socket int ris; // Raw input listening socket
int bos; // Beast output listening socket
int bis; // Beast input listening socket int bis; // Beast input listening socket
int https; // HTTP listening socket int https; // HTTP listening socket
char *rawOut; // Buffer for building raw output data
int rawOutUsed; // How much of the buffer is currently used struct net_writer raw_out; // Raw output
char *beastOut; // Buffer for building beast output data struct net_writer beast_out; // Beast-format output
int beastOutUsed; // How much if the buffer is currently used struct net_writer sbs_out; // SBS-format output
#ifdef _WIN32 #ifdef _WIN32
WSADATA wsaData; // Windows socket initialisation WSADATA wsaData; // Windows socket initialisation
#endif #endif
@ -295,12 +302,10 @@ struct { // Internal state
int debug; // Debugging mode int debug; // Debugging mode
int net; // Enable networking int net; // Enable networking
int net_only; // Enable just networking int net_only; // Enable just networking
int net_heartbeat_count; // TCP heartbeat counter int net_heartbeat_interval; // TCP heartbeat interval (seconds)
int net_heartbeat_rate; // TCP heartbeat rate
int net_output_sbs_port; // SBS output TCP port int net_output_sbs_port; // SBS output TCP port
int net_output_raw_size; // Minimum Size of the output raw data int net_output_flush_size; // Minimum Size of output data
int net_output_raw_rate; // Rate (in 64mS increments) of output raw data int net_output_flush_interval; // Maximum interval (in seconds) between outputwrites
int net_output_raw_rate_count; // Rate (in 64mS increments) of output raw data
int net_output_raw_port; // Raw output TCP port int net_output_raw_port; // Raw output TCP port
int net_input_raw_port; // Raw input TCP port int net_input_raw_port; // Raw input TCP port
int net_output_beast_port; // Beast output TCP port int net_output_beast_port; // Beast output TCP port
@ -348,9 +353,6 @@ struct { // Internal state
unsigned int stat_bit_fix[MODES_MAX_BITERRORS]; unsigned int stat_bit_fix[MODES_MAX_BITERRORS];
unsigned int stat_http_requests; unsigned int stat_http_requests;
unsigned int stat_sbs_connections;
unsigned int stat_raw_connections;
unsigned int stat_beast_connections;
unsigned int stat_out_of_phase; unsigned int stat_out_of_phase;
unsigned int stat_ph_demodulated0; unsigned int stat_ph_demodulated0;
unsigned int stat_ph_demodulated1; unsigned int stat_ph_demodulated1;
@ -453,11 +455,9 @@ struct stDF *interactiveFindDF (uint32_t addr);
// Functions exported from net_io.c // Functions exported from net_io.c
// //
void modesInitNet (void); void modesInitNet (void);
void modesReadFromClients (void);
void modesSendAllClients (int service, void *msg, int len);
void modesQueueOutput (struct modesMessage *mm); void modesQueueOutput (struct modesMessage *mm);
void modesReadFromClient(struct client *c, char *sep, int(*handler)(struct client *, char *)); void modesReadFromClient(struct client *c, char *sep, int(*handler)(struct client *, char *));
void modesNetCleanup (void); void modesNetPeriodicWork (void);
#ifdef __cplusplus #ifdef __cplusplus
} }

40
mode_s.c
View file

@ -1849,43 +1849,6 @@ void detectModeS(uint16_t *m, uint32_t mlen) {
use_correction = 0; use_correction = 0;
} }
} }
//Send any remaining partial raw buffers now
if (Modes.rawOutUsed || Modes.beastOutUsed)
{
Modes.net_output_raw_rate_count++;
if (Modes.net_output_raw_rate_count > Modes.net_output_raw_rate)
{
if (Modes.rawOutUsed) {
modesSendAllClients(Modes.ros, Modes.rawOut, Modes.rawOutUsed);
Modes.rawOutUsed = 0;
}
if (Modes.beastOutUsed) {
modesSendAllClients(Modes.bos, Modes.beastOut, Modes.beastOutUsed);
Modes.beastOutUsed = 0;
}
Modes.net_output_raw_rate_count = 0;
}
}
else if ( (Modes.net)
&& (Modes.net_heartbeat_rate)
&& ((++Modes.net_heartbeat_count) > Modes.net_heartbeat_rate) ) {
//
// We haven't received any Mode A/C/S messages for some time. To try and keep any TCP
// links alive, send a null frame. This will help stop any routers discarding our TCP
// link which will cause an un-recoverable link error if/when a real frame arrives.
//
// Fudge up a null message
memset(&mm, 0, sizeof(mm));
mm.msgbits = MODES_SHORT_MSG_BITS;
mm.timestampMsg = Modes.timestampBlk;
// Feed output clients
modesQueueOutput(&mm);
// Reset the heartbeat counter
Modes.net_heartbeat_count = 0;
}
} }
// //
//========================================================================= //=========================================================================
@ -1910,9 +1873,6 @@ void useModesMessage(struct modesMessage *mm) {
// Feed output clients // Feed output clients
if (Modes.net) {modesQueueOutput(mm);} if (Modes.net) {modesQueueOutput(mm);}
// Heartbeat not required whilst we're seeing real messages
Modes.net_heartbeat_count = 0;
} }
} }
// //

216
net_io.c
View file

@ -49,6 +49,7 @@
struct service { struct service {
char *descr; char *descr;
int *socket; int *socket;
struct net_writer *writer;
int port; int port;
int enabled; int enabled;
}; };
@ -59,12 +60,12 @@ void modesInitNet(void) {
int j; int j;
struct service svc[MODES_NET_SERVICES_NUM] = { struct service svc[MODES_NET_SERVICES_NUM] = {
{"Raw TCP output", &Modes.ros, Modes.net_output_raw_port, 1}, {"Raw TCP output", &Modes.raw_out.socket, &Modes.raw_out, Modes.net_output_raw_port, 1},
{"Raw TCP input", &Modes.ris, Modes.net_input_raw_port, 1}, {"Raw TCP input", &Modes.ris, NULL, Modes.net_input_raw_port, 1},
{"Beast TCP output", &Modes.bos, Modes.net_output_beast_port, 1}, {"Beast TCP output", &Modes.beast_out.socket, &Modes.beast_out, Modes.net_output_beast_port, 1},
{"Beast TCP input", &Modes.bis, Modes.net_input_beast_port, 1}, {"Beast TCP input", &Modes.bis, NULL, Modes.net_input_beast_port, 1},
{"HTTP server", &Modes.https, Modes.net_http_port, 1}, {"HTTP server", &Modes.https, NULL, Modes.net_http_port, 1},
{"Basestation TCP output", &Modes.sbsos, Modes.net_output_sbs_port, 1} {"Basestation TCP output", &Modes.sbs_out.socket, &Modes.sbs_out, Modes.net_output_sbs_port, 1}
}; };
memcpy(&services, &svc, sizeof(svc));//services = svc; memcpy(&services, &svc, sizeof(svc));//services = svc;
@ -93,6 +94,18 @@ void modesInitNet(void) {
} }
anetNonBlock(Modes.aneterr, s); anetNonBlock(Modes.aneterr, s);
*services[j].socket = s; *services[j].socket = s;
if (services[j].writer) {
if (! (services[j].writer->data = malloc(MODES_OUT_BUF_SIZE)) ) {
fprintf(stderr, "Out of memory allocating output buffer for service %s\n", services[j].descr);
exit(1);
}
services[j].writer->socket = s;
services[j].writer->connections = 0;
services[j].writer->dataUsed = 0;
services[j].writer->lastWrite = time(NULL);
}
} else { } else {
if (Modes.debug & MODES_DEBUG_NET) printf("%s port is disabled\n", services[j].descr); if (Modes.debug & MODES_DEBUG_NET) printf("%s port is disabled\n", services[j].descr);
} }
@ -127,9 +140,11 @@ struct client * modesAcceptClients(void) {
Modes.clients = c; Modes.clients = c;
anetSetSendBuffer(Modes.aneterr,fd, (MODES_NET_SNDBUF_SIZE << Modes.net_sndbuf_size)); anetSetSendBuffer(Modes.aneterr,fd, (MODES_NET_SNDBUF_SIZE << Modes.net_sndbuf_size));
if (*services[j].socket == Modes.sbsos) Modes.stat_sbs_connections++; if (services[j].writer) {
if (*services[j].socket == Modes.ros) Modes.stat_raw_connections++; if (++ services[j].writer->connections == 1) {
if (*services[j].socket == Modes.bos) Modes.stat_beast_connections++; services[j].writer->lastWrite = time(NULL); // suppress heartbeat initially
}
}
j--; // Try again with the same listening port j--; // Try again with the same listening port
@ -145,18 +160,21 @@ struct client * modesAcceptClients(void) {
// On error free the client, collect the structure, adjust maxfd if needed. // On error free the client, collect the structure, adjust maxfd if needed.
// //
void modesFreeClient(struct client *c) { void modesFreeClient(struct client *c) {
int j;
// Clean up, but defer removing from the list until modesNetCleanup(). // Clean up, but defer removing from the list until modesNetCleanup().
// This is because there may be stackframes still pointing at this // This is because there may be stackframes still pointing at this
// client (unpredictably: reading from client A may cause client B to // client (unpredictably: reading from client A may cause client B to
// be freed) // be freed)
close(c->fd); close(c->fd);
if (c->service == Modes.sbsos) {
if (Modes.stat_sbs_connections) Modes.stat_sbs_connections--; for (j = 0; j < MODES_NET_SERVICES_NUM; j++) {
} else if (c->service == Modes.ros) { if (c->service == *services[j].socket) {
if (Modes.stat_raw_connections) Modes.stat_raw_connections--; if (services[j].writer)
} else if (c->service == Modes.bos) { services[j].writer->connections--;
if (Modes.stat_beast_connections) Modes.stat_beast_connections--; break;
}
} }
if (Modes.debug & MODES_DEBUG_NET) if (Modes.debug & MODES_DEBUG_NET)
@ -169,36 +187,72 @@ void modesFreeClient(struct client *c) {
// //
//========================================================================= //=========================================================================
// //
// Send the specified message to all clients listening for a given service // Send the write buffer for the specified writer to all connected clients
// //
void modesSendAllClients(int service, void *msg, int len) { static void flushWrites(struct net_writer *writer) {
struct client *c; struct client *c;
for (c = Modes.clients; c; c = c->next) { for (c = Modes.clients; c; c = c->next) {
if (c->service == service) { if (c->service == writer->socket) {
#ifndef _WIN32 #ifndef _WIN32
int nwritten = write(c->fd, msg, len); int nwritten = write(c->fd, writer->data, writer->dataUsed);
#else #else
int nwritten = send(c->fd, msg, len, 0 ); int nwritten = send(c->fd, writer->data, writer->dataUsed, 0 );
#endif #endif
if (nwritten != len) { if (nwritten != writer->dataUsed) {
modesFreeClient(c); modesFreeClient(c);
} }
} }
} }
writer->dataUsed = 0;
writer->lastWrite = time(NULL);
} }
// Prepare to write up to 'len' bytes to the given net_writer.
// Returns a pointer to write to, or NULL to skip this write.
static void *prepareWrite(struct net_writer *writer, int len) {
if (!writer ||
!writer->connections ||
!writer->data)
return NULL;
if (len >= MODES_OUT_BUF_SIZE)
return NULL;
if (writer->dataUsed + len >= MODES_OUT_BUF_SIZE) {
// Flush now to free some space
flushWrites(writer);
}
return writer->data + writer->dataUsed;
}
// Complete a write previously begun by prepareWrite.
// endptr should point one byte past the last byte written
// to the buffer returned from prepareWrite.
static void completeWrite(struct net_writer *writer, void *endptr) {
writer->dataUsed = endptr - writer->data;
if (writer->dataUsed >= Modes.net_output_flush_size) {
flushWrites(writer);
}
}
// //
//========================================================================= //=========================================================================
// //
// Write raw output in Beast Binary format with Timestamp to TCP clients // Write raw output in Beast Binary format with Timestamp to TCP clients
// //
void modesSendBeastOutput(struct modesMessage *mm) { void modesSendBeastOutput(struct modesMessage *mm) {
char *p = &Modes.beastOut[Modes.beastOutUsed];
int msgLen = mm->msgbits / 8; int msgLen = mm->msgbits / 8;
char *p = prepareWrite(&Modes.beast_out, 2 + 2 * (7 + msgLen));
char * pTimeStamp; char * pTimeStamp;
char ch; char ch;
int j; int j;
int iOutLen = msgLen + 9; // Escape, msgtype, timestamp, sigLevel, msg
if (!p)
return;
*p++ = 0x1a; *p++ = 0x1a;
if (msgLen == MODES_SHORT_MSG_BYTES) if (msgLen == MODES_SHORT_MSG_BYTES)
@ -213,36 +267,34 @@ void modesSendBeastOutput(struct modesMessage *mm) {
pTimeStamp = (char *) &mm->timestampMsg; pTimeStamp = (char *) &mm->timestampMsg;
for (j = 5; j >= 0; j--) { for (j = 5; j >= 0; j--) {
*p++ = (ch = pTimeStamp[j]); *p++ = (ch = pTimeStamp[j]);
if (0x1A == ch) {*p++ = ch; iOutLen++;} if (0x1A == ch) {*p++ = ch; }
} }
*p++ = (ch = mm->signalLevel); *p++ = (ch = mm->signalLevel);
if (0x1A == ch) {*p++ = ch; iOutLen++;} if (0x1A == ch) {*p++ = ch; }
for (j = 0; j < msgLen; j++) { for (j = 0; j < msgLen; j++) {
*p++ = (ch = mm->msg[j]); *p++ = (ch = mm->msg[j]);
if (0x1A == ch) {*p++ = ch; iOutLen++;} if (0x1A == ch) {*p++ = ch; }
} }
Modes.beastOutUsed += iOutLen; completeWrite(&Modes.beast_out, p);
if (Modes.beastOutUsed >= Modes.net_output_raw_size)
{
modesSendAllClients(Modes.bos, Modes.beastOut, Modes.beastOutUsed);
Modes.beastOutUsed = 0;
Modes.net_output_raw_rate_count = 0;
}
} }
// //
//========================================================================= //=========================================================================
// //
// Write raw output to TCP clients // Write raw output to TCP clients
// //
void modesSendRawOutput(struct modesMessage *mm) { void modesSendRawOutput(struct modesMessage *mm) {
char *p = &Modes.rawOut[Modes.rawOutUsed];
int msgLen = mm->msgbits / 8; int msgLen = mm->msgbits / 8;
char *p = prepareWrite(&Modes.raw_out, msgLen*2 + 15);
int j; int j;
unsigned char * pTimeStamp; unsigned char * pTimeStamp;
if (!p)
return;
if (Modes.mlat && mm->timestampMsg) { if (Modes.mlat && mm->timestampMsg) {
*p++ = '@'; *p++ = '@';
pTimeStamp = (unsigned char *) &mm->timestampMsg; pTimeStamp = (unsigned char *) &mm->timestampMsg;
@ -250,7 +302,6 @@ void modesSendRawOutput(struct modesMessage *mm) {
sprintf(p, "%02X", pTimeStamp[j]); sprintf(p, "%02X", pTimeStamp[j]);
p += 2; p += 2;
} }
Modes.rawOutUsed += 12; // additional 12 characters for timestamp
} else } else
*p++ = '*'; *p++ = '*';
@ -262,13 +313,7 @@ void modesSendRawOutput(struct modesMessage *mm) {
*p++ = ';'; *p++ = ';';
*p++ = '\n'; *p++ = '\n';
Modes.rawOutUsed += ((msgLen*2) + 3); completeWrite(&Modes.raw_out, p);
if (Modes.rawOutUsed >= Modes.net_output_raw_size)
{
modesSendAllClients(Modes.ros, Modes.rawOut, Modes.rawOutUsed);
Modes.rawOutUsed = 0;
Modes.net_output_raw_rate_count = 0;
}
} }
// //
//========================================================================= //=========================================================================
@ -277,17 +322,27 @@ void modesSendRawOutput(struct modesMessage *mm) {
// The message structure mm->bFlags tells us what has been updated by this message // The message structure mm->bFlags tells us what has been updated by this message
// //
void modesSendSBSOutput(struct modesMessage *mm) { void modesSendSBSOutput(struct modesMessage *mm) {
char msg[256], *p = msg; char *p = prepareWrite(&Modes.sbs_out, 200);
uint32_t offset; uint32_t offset;
struct timeb epocTime_receive, epocTime_now; struct timeb epocTime_receive, epocTime_now;
struct tm stTime_receive, stTime_now; struct tm stTime_receive, stTime_now;
int msgType; int msgType;
if (!p)
return;
// //
// SBS BS style output checked against the following reference // SBS BS style output checked against the following reference
// http://www.homepages.mcb.net/bones/SBS/Article/Barebones42_Socket_Data.htm - seems comprehensive // http://www.homepages.mcb.net/bones/SBS/Article/Barebones42_Socket_Data.htm - seems comprehensive
// //
if (mm->msgtype == -1) {
// heartbeat
p += sprintf(p, "\r\n");
completeWrite(&Modes.sbs_out, p);
return;
}
// Decide on the basic SBS Message Type // Decide on the basic SBS Message Type
if ((mm->msgtype == 4) || (mm->msgtype == 20)) { if ((mm->msgtype == 4) || (mm->msgtype == 20)) {
msgType = 5; msgType = 5;
@ -434,15 +489,16 @@ void modesSendSBSOutput(struct modesMessage *mm) {
} }
p += sprintf(p, "\r\n"); p += sprintf(p, "\r\n");
modesSendAllClients(Modes.sbsos, msg, p-msg);
completeWrite(&Modes.sbs_out, p);
} }
// //
//========================================================================= //=========================================================================
// //
void modesQueueOutput(struct modesMessage *mm) { void modesQueueOutput(struct modesMessage *mm) {
if (Modes.stat_sbs_connections) {modesSendSBSOutput(mm);} modesSendSBSOutput(mm);
if (Modes.stat_beast_connections) {modesSendBeastOutput(mm);} modesSendBeastOutput(mm);
if (Modes.stat_raw_connections) {modesSendRawOutput(mm);} modesSendRawOutput(mm);
} }
// //
//========================================================================= //=========================================================================
@ -928,17 +984,21 @@ void modesReadFromClient(struct client *c, char *sep,
} }
} }
} }
//
//=========================================================================
//
// Read data from clients. This function actually delegates a lower-level
// function that depends on the kind of service (raw, http, ...).
//
void modesReadFromClients(void) {
struct client *c = modesAcceptClients(); //
// Perform periodic network work
//
void modesNetPeriodicWork(void) {
struct client *c, **prev;
time_t now = time(NULL);
int j;
int need_heartbeat = 0, need_flush = 0;
for (; c; c = c->next) { // Accept new connetions
modesAcceptClients();
// Read from clients
for (c = Modes.clients; c; c = c->next) {
if (c->service == Modes.ris) { if (c->service == Modes.ris) {
modesReadFromClient(c,"\n",decodeHexMessage); modesReadFromClient(c,"\n",decodeHexMessage);
} else if (c->service == Modes.bis) { } else if (c->service == Modes.bis) {
@ -947,14 +1007,52 @@ void modesReadFromClients(void) {
modesReadFromClient(c,"\r\n\r\n",handleHTTPRequest); modesReadFromClient(c,"\r\n\r\n",handleHTTPRequest);
} }
} }
// If we have generated no messages for a while, generate
// a dummy heartbeat message.
if (Modes.net_heartbeat_interval) {
for (j = 0; j < MODES_NET_SERVICES_NUM; j++) {
if (services[j].writer &&
services[j].writer->connections &&
(services[j].writer->lastWrite + Modes.net_heartbeat_interval) <= now) {
need_flush = 1;
if (services[j].writer->dataUsed == 0) {
need_heartbeat = 1;
break;
}
}
}
} }
if (need_heartbeat) {
// //
// Perform cleanup of any recently-closed clients. // We haven't sent any traffic for some time. To try and keep any TCP
// links alive, send a null frame. This will help stop any routers discarding our TCP
// link which will cause an un-recoverable link error if/when a real frame arrives.
// //
void modesNetCleanup(void) { // Fudge up a null message
struct client *c, **prev; struct modesMessage mm;
memset(&mm, 0, sizeof(mm));
mm.msgbits = MODES_SHORT_MSG_BITS;
mm.timestampMsg = 0;
mm.msgtype = -1;
// Feed output clients
modesQueueOutput(&mm);
}
// If we have data that has been waiting to be written for a while,
// write it now.
for (j = 0; j < MODES_NET_SERVICES_NUM; j++) {
if (services[j].writer &&
services[j].writer->dataUsed &&
(need_flush || (services[j].writer->lastWrite + Modes.net_output_flush_interval) <= now)) {
flushWrites(services[j].writer);
}
}
// Unlink and free closed clients
for (prev = &Modes.clients, c = *prev; c; c = *prev) { for (prev = &Modes.clients, c = *prev; c; c = *prev) {
if (c->fd == -1) { if (c->fd == -1) {
// Recently closed, prune from list // Recently closed, prune from list