eddy/small_tel
1
0
Fork 0
mirror of https://github.com/eddyem/small_tel.git synced 2026-03-20 08:41:03 +03:00
Code Issues Packages Projects Releases Wiki Activity

Change netdaemon to poll & work with libusefull_macros, add weatherdaemon

Browse Source
This commit is contained in:
Edward Emelianov
2021-01-02 20:04:20 +03:00
parent dff0f9e43f
commit 0028ba8a4b
45 changed files with 1835 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

278
Daemons/netdaemon/socket.c
View File

@@ -20,19 +20,24 @@
* MA 02110-1301, USA.
*
*/
#include "usefull_macros.h"
#include "socket.h"
#include "term.h"
#include <netdb.h> // addrinfo
#include <arpa/inet.h> // inet_ntop
#include <limits.h> // INT_xxx
#include <signal.h> // pthread_kill
#include <unistd.h> // daemon
#include <poll.h> // poll
#include <pthread.h>
#include <signal.h> // pthread_kill
#include <stdio.h>
#include <string.h>
#include <sys/syscall.h> // syscall
#include <unistd.h> // daemon
#include <usefull_macros.h>
#include "cmdlnopts.h" // glob_pars
#include "socket.h"
#include "term.h"
#define BUFLEN (10240)
// temporary buffers
#define BUFLEN (1024)
// Max amount of connections
#define BACKLOG (30)
@@ -42,37 +47,7 @@ extern glob_pars *GP;
* Define global data buffers here
*/
/**************** COMMON FUNCTIONS ****************/
/**
* wait for answer from socket
* @param sock - socket fd
* @return 0 in case of error or timeout, 1 in case of socket ready
*/
static int waittoread(int sock){
fd_set fds;
struct timeval timeout;
int rc;
timeout.tv_sec = 1; // wait not more than 1 second
timeout.tv_usec = 0;
FD_ZERO(&fds);
FD_SET(sock, &fds);
do{
rc = select(sock+1, &fds, NULL, NULL, &timeout);
if(rc < 0){
if(errno != EINTR){
WARN("select()");
return 0;
}
continue;
}
break;
}while(1);
if(FD_ISSET(sock, &fds)) return 1;
return 0;
}
/**************** SERVER FUNCTIONS ****************/
static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
/**
* Send data over socket
* @param sock - socket fd
@@ -94,9 +69,11 @@ static int send_data(int sock, int webquery, char *textbuf){
"Content-type: text/plain\r\nContent-Length: %zd\r\n\r\n", Len);
if(L < 0){
WARN("sprintf()");
LOGWARN("sprintf()");
return 0;
}
if(L != write(sock, tbuf, L)){
LOGWARN("Can't write header");
WARN("write");
return 0;
}
@@ -105,121 +82,162 @@ static int send_data(int sock, int webquery, char *textbuf){
//DBG("send %zd bytes\nBUF: %s", Len, buf);
if(Len != write(sock, textbuf, Len)){
WARN("write()");
LOGERR("send_data(): write() failed");
return 0;
}
LOGDBG("fd %d, write %s", textbuf);
return 1;
}
// search a first word after needle without spaces
static char* stringscan(char *str, char *needle){
char *a, *e;
char *a;//, *e;
char *end = str + strlen(str);
a = strstr(str, needle);
if(!a) return NULL;
a += strlen(needle);
while (a < end && (*a == ' ' || *a == '\r' || *a == '\t' || *a == '\r')) a++;
if(a >= end) return NULL;
e = strchr(a, ' ');
if(e) *e = 0;
return a;
}
static void *handle_socket(void *asock){
//LOG("handle_socket(): getpid: %d, pthread_self: %lu, tid: %lu",getpid(), pthread_self(), syscall(SYS_gettid));
/**
* @brief handle_socket - read information from socket
* @param sock - socket fd
* @param chkheader - ==1 on first run
* @return 1 if socket closed
*/
static int handle_socket(int sock, int notchkhdr){
FNAME();
int sock = *((int*)asock);
int webquery = 0; // whether query is web or regular
char buff[BUFLEN];
ssize_t rd;
double t0 = dtime();
/*
* INSERT CODE HERE
* change to while(1) if socket shouldn't be closed after data transmission
*/
while(dtime() - t0 < SOCKET_TIMEOUT){
if(!waittoread(sock)){ // no data incoming
continue;
}
if(!(rd = read(sock, buff, BUFLEN-1))){
//LOG("socket closed. Exit");
break;
}
//LOG("client send %zd bytes", rd);
DBG("Got %zd bytes", rd);
if(rd < 0){ // error
//LOG("some error occured");
DBG("Nothing to read from fd %d (ret: %zd)", sock, rd);
break;
}
// add trailing zero to be on the safe side
buff[rd] = 0;
// now we should check what do user want
char *got, *found = buff;
if((got = stringscan(buff, "GET")) || (got = stringscan(buff, "POST"))){ // web query
if(!(rd = read(sock, buff, BUFLEN-1))){
LOGMSG("Client %d closed", sock);
return 1;
}
//LOG("client send %zd bytes", rd);
DBG("Got %zd bytes", rd);
if(rd < 0){ // error
LOGWARN("Client %d close socket on error", sock);
DBG("Nothing to read from fd %d (ret: %zd)", sock, rd);
return 1;
}
// add trailing zero to be on the safe side
buff[rd] = 0;
// now we should check what do user want
char *found = buff;
DBG("user send: %s", buff);
if(!notchkhdr){
if(0 == strncmp(buff, "GET", 3)){
DBG("GET");
// GET web query have format GET /some.resource
webquery = 1;
char *slash = strchr(got, '/');
if(slash) found = slash + 1;
// web query have format GET /some.resource
}
// here we can process user data
DBG("user send: %s\nfound=%s", buff, found);
if(GP->echo){
if(!send_data(sock, webquery, found)){
LOG("can't send data, some error occured");
char *slash = strchr(buff, '/');
if(slash){
found = slash + 1;
char *eol = strstr(found, "HTTP");
if(eol) *eol = 0;
}
}else if(0 == strncmp(buff, "POST", 4)){
DBG("POST");
webquery = 1;
// search content length of POST query
char *cl = stringscan(buff, "Content-Length:");
if(cl){
int contlen = atoi(cl);
int l = strlen(buff);
if(contlen && l > contlen) found = &buff[l - contlen];
}
}
pthread_mutex_lock(&mutex);
/*
* INSERT CODE HERE
* Process user commands here & send him an answer
* remove trailing break if socket shouldn't be closed after server sent data
*/
pthread_mutex_unlock(&mutex);
break;
}
close(sock);
//DBG("closed");
//LOG("socket closed, exit");
pthread_exit(NULL);
return NULL;
// here we can process user data
DBG("found=%s", found);
LOGDBG("sockfd=%d, got %s", sock, buff);
if(GP->echo){
if(!send_data(sock, webquery, found)){
LOGWARN("Can't send data, some error occured");
return 1;
}
}
/*
*
* INSERT CODE HERE
* Process user commands here & send him an answer
* remove trailing break if socket shouldn't be closed after server sent data
*
*/
if(webquery) return 1; // close web query after message processing
return 0;
}
// main socket server
static void *server(void *asock){
LOG("server(): getpid: %d, pthread_self: %lu, tid: %lu",getpid(), pthread_self(), syscall(SYS_gettid));
LOGMSG("server()");
int sock = *((int*)asock);
if(listen(sock, BACKLOG) == -1){
LOG("listen() failed");
LOGERR("listen() failed");
WARN("listen");
return NULL;
}
int nfd = 1; // current fd amount in poll_set
struct pollfd poll_set[MAX_FDS];
int notchkhdr[MAX_FDS];
memset(poll_set, 0, sizeof(poll_set));
memset(notchkhdr, 0, sizeof(notchkhdr));
poll_set[0].fd = sock;
poll_set[0].events = POLLIN;
while(1){
socklen_t size = sizeof(struct sockaddr_in);
struct sockaddr_in their_addr;
int newsock;
if(!waittoread(sock)) continue;
newsock = accept(sock, (struct sockaddr*)&their_addr, &size);
if(newsock <= 0){
LOG("accept() failed");
WARN("accept()");
continue;
}
struct sockaddr_in* pV4Addr = (struct sockaddr_in*)&their_addr;
struct in_addr ipAddr = pV4Addr->sin_addr;
char str[INET_ADDRSTRLEN];
inet_ntop(AF_INET, &ipAddr, str, INET_ADDRSTRLEN);
//LOG("get connection from %s", str);
DBG("Got connection from %s\n", str);
pthread_t handler_thread;
if(pthread_create(&handler_thread, NULL, handle_socket, (void*) &newsock)){
LOG("server(): pthread_create() failed");
WARN("pthread_create()");
}else{
DBG("Thread created, detouch");
pthread_detach(handler_thread); // don't care about thread state
}
poll(poll_set, nfd, 1); // poll for 1ms
for(int fdidx = 0; fdidx < nfd; ++fdidx){ // poll opened FDs
if((poll_set[fdidx].revents & POLLIN) == 0) continue;
poll_set[fdidx].revents = 0;
if(fdidx){ // client
int fd = poll_set[fdidx].fd;
if(handle_socket(fd, notchkhdr[fdidx])){ // socket closed - remove it from list
close(fd);
DBG("Client with fd %d closed", fd);
LOGMSG("Client %d disconnected", fd);
// move last to free space
poll_set[fdidx] = poll_set[nfd - 1];
notchkhdr[fdidx] = notchkhdr[nfd - 1];
--nfd;
}else notchkhdr[fdidx] = 1;
}else{ // server
socklen_t size = sizeof(struct sockaddr_in);
struct sockaddr_in their_addr;
int newsock = accept(sock, (struct sockaddr*)&their_addr, &size);
if(newsock <= 0){
LOGERR("server(): accept() failed");
WARN("accept()");
continue;
}
struct in_addr ipAddr = their_addr.sin_addr;
char str[INET_ADDRSTRLEN];
inet_ntop(AF_INET, &ipAddr, str, INET_ADDRSTRLEN);
DBG("Connection from %s, give fd=%d", str, newsock);
LOGMSG("Got connection from %s, fd=%d", str, newsock);
if(nfd == MAX_FDS){
LOGWARN("Max amount of connections: disconnect %s (%d)", str, newsock);
send_data(newsock, 0, "Max amount of connections reached!\n");
WARNX("Limit of connections reached");
close(newsock);
}else{
memset(&poll_set[nfd], 0, sizeof(struct pollfd));
poll_set[nfd].fd = newsock;
poll_set[nfd].events = POLLIN;
notchkhdr[nfd] = 0;
++nfd;
}
}
} // endfor
/*
* INSERT CODE HERE
* Send broadcast messages
*/
}
LOG("server(): UNREACHABLE CODE REACHED!");
LOGERR("server(): UNREACHABLE CODE REACHED!");
}
// data gathering & socket management
@@ -227,36 +245,26 @@ static void daemon_(int sock){
if(sock < 0) return;
pthread_t sock_thread;
if(pthread_create(&sock_thread, NULL, server, (void*) &sock)){
LOG("daemon_(): pthread_create() failed");
LOGERR("daemon_(): pthread_create() failed");
ERR("pthread_create()");
}
double tgot = 0.;
do{
if(pthread_kill(sock_thread, 0) == ESRCH){ // died
WARNX("Sockets thread died");
LOG("Sockets thread died");
LOGWARN("Sockets thread died");
pthread_join(sock_thread, NULL);
if(pthread_create(&sock_thread, NULL, server, (void*) &sock)){
LOG("daemon_(): new pthread_create() failed");
LOGERR("daemon_(): new pthread_create() failed");
ERR("pthread_create()");
}
}
usleep(1000); // sleep a little or thread's won't be able to lock mutex
if(dtime() - tgot < T_INTERVAL) continue;
tgot = dtime();
usleep(1000); // sleep a little
/*
* INSERT CODE HERE
* Gather data (poll_device)
*/
// copy temporary buffers to main
pthread_mutex_lock(&mutex);
/*
* INSERT CODE HERE
* fill global data buffers
*/
pthread_mutex_unlock(&mutex);
}while(1);
LOG("daemon_(): UNREACHABLE CODE REACHED!");
LOGERR("daemon_(): UNREACHABLE CODE REACHED!");
}
/**
@@ -270,7 +278,9 @@ void daemonize(char *port){
hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM;
hints.ai_flags = AI_PASSIVE;
// To accept only local sockets replace NULL with "127.0.0.1" and remove AI_PASSIVE
if(getaddrinfo(NULL, port, &hints, &res) != 0){
LOGERR("getaddrinfo");
ERR("getaddrinfo");
}
struct sockaddr_in *ia = (struct sockaddr_in*)res->ai_addr;
@@ -284,24 +294,26 @@ void daemonize(char *port){
}
int reuseaddr = 1;
if(setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuseaddr, sizeof(int)) == -1){
LOGERR("setsockopt() error");
ERR("setsockopt");
}
if(bind(sock, p->ai_addr, p->ai_addrlen) == -1){
close(sock);
WARN("bind");
LOGWARN("bind() error");
continue;
}
break; // if we get here, we have a successfull connection
}
if(p == NULL){
LOG("failed to bind socket, exit");
LOGERR("daemonize(): failed to bind socket, exit");
// looped off the end of the list with no successful bind
ERRX("failed to bind socket");
}
freeaddrinfo(res);
daemon_(sock);
close(sock);
LOG("socket closed, exit");
LOGERR("daemonize(): UNREACHABLE CODE REACHED!");
signals(0);
}