Commit 86cf9c75 authored by Vladislav Rykov's avatar Vladislav Rykov
Browse files

restructuring project, new makefile

parent 51130f1d
......@@ -4,3 +4,6 @@
[submodule "lib/base64"]
path = lib/base64
url = https://github.com/zhicheng/base64.git
[submodule "json-parser"]
path = json-parser
url = https://github.com/udp/json-parser
CC = gcc
INCLUDES = -I/usr/include/postgresql -I$(LDIR)/gateway_protocol -I$(LDIR)/base64 -I$(LDIR)/task_queue
LFLAGS = -Llib
LDIR = lib
LIBD = -pthread -lpq
SRC = src/gateway.c
OBJ = obj
MAIN = gateway
.PHONY: depend clean
$(LDIR)/gateway_protocol/gateway_protocol.o:
$(CC) -c $(LDIR)/gateway_protocol/gateway_protocol.c -o $(OBJ)/gateway_protocol.o -I$(LDIR)/gateway_protocol
$(CC) -c $(LDIR)/base64/base64.c -o $(OBJ)/base64.o -I$(LDIR)/base64
$(CC) -c $(LDIR)/task_queue/task_queue.c -o $(OBJ)/task_queue.o -I$(LDIR)/task_queue
$(CC) $(SRC) $(OBJ)/gateway_protocol.o $(OBJ)/base64.o $(OBJ)/task_queue.o -o $(MAIN) $(LIBD) $(INCLUDES)
all: $(MAIN)
@echo Compiling gateway project
$(MAIN): $(OBJ)/gateway_protocol.o $(OBJ)/base64.o $(OBJ)/task_queue.o
$(CC) $(CFLAGS) $(INCLUDES) -o $(MAIN) $(OBJS) $(LFLAGS)
.c.o:
$(CC) $(CFLAGS) $(INCLUDES) -c $< -o $@
clean:
$(RM) *.o *~ $(MAIN)
depend: $(SRCS)
makedepend $(INCLUDES) $^
# line needed by makedepend
File added
File deleted
{
"gateway_id" : "fa:16:3e:8c:4c:ef",
"gateway_password" : "12345",
"db_type" : "PostgreSQL",
"db_connection" : {
"db_address" : "127.0.0.1",
"db_port" : "5432",
"db_name" : "iotserver",
"user_name" : "vlad",
"user_password" : "dev"
}
}
#ifndef BASE64_H
#define BASE64_H
#define BASE64_ENCODE_OUT_SIZE(s) ((unsigned int)((((s) + 2) / 3) * 4 + 1))
#define BASE64_DECODE_OUT_SIZE(s) ((unsigned int)(((s) / 4) * 3))
/*
* out is null-terminated encode string.
* return values is out length, exclusive terminating `\0'
*/
unsigned int
base64_encode(const unsigned char *in, unsigned int inlen, char *out);
/*
* return values is out length
*/
unsigned int
base64_decode(const char *in, unsigned int inlen, unsigned char *out);
#endif /* BASE64_H */
#ifndef __GATEWAY_PROTOCOL_H__
#define __GATEWAY_PROTOCOL_H__
#include <stdint.h>
#include <string.h>
#define GATEWAY_PROTOCOL_PACKET_SIZE_MAX 128
#ifdef __cplusplus
extern "C" {
#endif
typedef enum {
GATEWAY_PROTOCOL_PACKET_TYPE_DATA_SEND = 0x00,
GATEWAY_PROTOCOL_PACKET_TYPE_PEND_REQ = 0x04,
GATEWAY_PROTOCOL_PACKET_TYPE_PEND_SEND = 0x05,
GATEWAY_PROTOCOL_PACKET_TYPE_STAT = 0x10,
GATEWAY_PROTOCOL_PACKET_TYPE_TIME_REQ = 0x20,
GATEWAY_PROTOCOL_PACKET_TYPE_TIME_SEND = 0x21,
GATEWAY_PROTOCOL_PACKET_TYPE_UNKNOWN = 0xFF
} gateway_protocol_packet_type_t;
typedef enum {
GATEWAY_PROTOCOL_STAT_ACK = 0,
GATEWAY_PROTOCOL_STAT_ACK_PEND,
GATEWAY_PROTOCOL_STAT_NACK = 0xFF
} gateway_protocol_stat_t;
void gateway_protocol_init(const uint8_t *appkey, const uint8_t devid);
void gateway_protocol_packet_encode (
const gateway_protocol_packet_type_t packet_type,
const uint8_t payload_length,
const uint8_t *payload,
uint8_t *packet_length,
uint8_t *packet);
uint8_t gateway_protocol_packet_decode (
gateway_protocol_packet_type_t *packet_type,
uint8_t *payload_length,
uint8_t *payload,
const uint8_t packet_length,
const uint8_t *packet);
#ifdef __cplusplus
}
#endif
#endif // __GATEWAY_PROTOCOL_H__
\ No newline at end of file
/* vim: set et ts=3 sw=3 sts=3 ft=c:
*
* Copyright (C) 2012, 2013, 2014 James McLaughlin et al. All rights reserved.
* https://github.com/udp/json-parser
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
* 1. Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
* FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
* OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
* HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
* LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
* OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
* SUCH DAMAGE.
*/
#ifndef _JSON_H
#define _JSON_H
#ifndef json_char
#define json_char char
#endif
#ifndef json_int_t
#ifndef _MSC_VER
#include <inttypes.h>
#define json_int_t int64_t
#else
#define json_int_t __int64
#endif
#endif
#include <stdlib.h>
#ifdef __cplusplus
#include <string.h>
extern "C"
{
#endif
typedef struct
{
unsigned long max_memory;
int settings;
/* Custom allocator support (leave null to use malloc/free)
*/
void * (* mem_alloc) (size_t, int zero, void * user_data);
void (* mem_free) (void *, void * user_data);
void * user_data; /* will be passed to mem_alloc and mem_free */
size_t value_extra; /* how much extra space to allocate for values? */
} json_settings;
#define json_enable_comments 0x01
typedef enum
{
json_none,
json_object,
json_array,
json_integer,
json_double,
json_string,
json_boolean,
json_null
} json_type;
extern const struct _json_value json_value_none;
typedef struct _json_object_entry
{
json_char * name;
unsigned int name_length;
struct _json_value * value;
} json_object_entry;
typedef struct _json_value
{
struct _json_value * parent;
json_type type;
union
{
int boolean;
json_int_t integer;
double dbl;
struct
{
unsigned int length;
json_char * ptr; /* null terminated */
} string;
struct
{
unsigned int length;
json_object_entry * values;
#if defined(__cplusplus) && __cplusplus >= 201103L
decltype(values) begin () const
{ return values;
}
decltype(values) end () const
{ return values + length;
}
#endif
} object;
struct
{
unsigned int length;
struct _json_value ** values;
#if defined(__cplusplus) && __cplusplus >= 201103L
decltype(values) begin () const
{ return values;
}
decltype(values) end () const
{ return values + length;
}
#endif
} array;
} u;
union
{
struct _json_value * next_alloc;
void * object_mem;
} _reserved;
#ifdef JSON_TRACK_SOURCE
/* Location of the value in the source JSON
*/
unsigned int line, col;
#endif
/* Some C++ operator sugar */
#ifdef __cplusplus
public:
inline _json_value ()
{ memset (this, 0, sizeof (_json_value));
}
inline const struct _json_value &operator [] (int index) const
{
if (type != json_array || index < 0
|| ((unsigned int) index) >= u.array.length)
{
return json_value_none;
}
return *u.array.values [index];
}
inline const struct _json_value &operator [] (const char * index) const
{
if (type != json_object)
return json_value_none;
for (unsigned int i = 0; i < u.object.length; ++ i)
if (!strcmp (u.object.values [i].name, index))
return *u.object.values [i].value;
return json_value_none;
}
inline operator const char * () const
{
switch (type)
{
case json_string:
return u.string.ptr;
default:
return "";
};
}
inline operator json_int_t () const
{
switch (type)
{
case json_integer:
return u.integer;
case json_double:
return (json_int_t) u.dbl;
default:
return 0;
};
}
inline operator bool () const
{
if (type != json_boolean)
return false;
return u.boolean != 0;
}
inline operator double () const
{
switch (type)
{
case json_integer:
return (double) u.integer;
case json_double:
return u.dbl;
default:
return 0;
};
}
#endif
} json_value;
json_value * json_parse (const json_char * json,
size_t length);
#define json_error_max 128
json_value * json_parse_ex (json_settings * settings,
const json_char * json,
size_t length,
char * error);
void json_value_free (json_value *);
/* Not usually necessary, unless you used a custom mem_alloc and now want to
* use a custom mem_free.
*/
void json_value_free_ex (json_settings * settings,
json_value *);
#ifdef __cplusplus
} /* extern "C" */
#endif
#endif
#ifndef __TASK_QUEUE_H__
#define __TASK_QUEUE_H__
/* Task queue that create threads for completing tasks
* having an upper threads limit
*/
struct task_queue;
typedef struct task_queue task_queue_t;
typedef void (*task_func_t)(void *arg);
task_queue_t * task_queue_create(const int max_threads);
void task_queue_destroy(task_queue_t *tq);
int task_queue_enqueue(task_queue_t *tq, task_func_t task, void *arg);
void task_queue_suspend(task_queue_t *tq);
void task_queue_unsuspend(task_queue_t *tq);
int task_queue_get_size(task_queue_t *tq);
int task_queue_is_empty(task_queue_t *tq);
#endif // __TASK_QUEUE_H__
[Unit]
Description=HPC&A IoT Gateway
[Service]
User=vlad
Type=simple
ExecStart=/home/vlad/thso.gateway/bin/gateway 1>/dev/null
WorkingDirectory=/home/vlad/thso.gateway
Restart=on-failure
KillSignal=SIGKILL
[Install]
WantedBy=multi-user.target
json-parser @ e6426aef
Subproject commit e6426aefb6ded8dcc549d847e2a453aea5dd04a6
CC = gcc
CFLAGS = -Wall
LFLAGS = -pthread -lpq -lm
INC_DIR = ../inc
OBJ_DIR = ../obj
BIN_DIR = ../bin
SRC_DIR = .
INCLUDES = -I/usr/include/postgresql \
-I$(INC_DIR)
TARGET = $(BIN_DIR)/gateway
OBJS = $(wildcard $(OBJ_DIR)/*.o)
#
#SRCS = $(wildcard $(SRC_DIR)/*.c)
#OBJS = $(SRCS:.c=.o)
#
$(TARGET) : $(OBJS)
$(CC) $(CFLAGS) $(INCLUDES) $(OBJS) -o $(TARGET) $(LFLAGS)
$(OBJ_DIR)/%.o : $(SRC_DIR)/%.c
$(CC) -c -MD $(CFLAGS) $(INCLUDES) $< -o $@
-include $(OBJ_DIR)/*.d
.PHONY: clean print
clean :
rm -f $(BIN_DIR)/* $(OBJ_DIR)/*
print :
$(info $$OBJS is [${OBJS}])
/* This is a public domain base64 implementation written by WEI Zhicheng. */
#include "base64.h"
#define BASE64_PAD '='
#define BASE64DE_FIRST '+'
#define BASE64DE_LAST 'z'
/* BASE 64 encode table */
static const char base64en[] = {
'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H',
'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P',
'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X',
'Y', 'Z', 'a', 'b', 'c', 'd', 'e', 'f',
'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n',
'o', 'p', 'q', 'r', 's', 't', 'u', 'v',
'w', 'x', 'y', 'z', '0', '1', '2', '3',
'4', '5', '6', '7', '8', '9', '+', '/',
};
/* ASCII order for BASE 64 decode, 255 in unused character */
static const unsigned char base64de[] = {
/* nul, soh, stx, etx, eot, enq, ack, bel, */
255, 255, 255, 255, 255, 255, 255, 255,
/* bs, ht, nl, vt, np, cr, so, si, */
255, 255, 255, 255, 255, 255, 255, 255,
/* dle, dc1, dc2, dc3, dc4, nak, syn, etb, */
255, 255, 255, 255, 255, 255, 255, 255,
/* can, em, sub, esc, fs, gs, rs, us, */
255, 255, 255, 255, 255, 255, 255, 255,
/* sp, '!', '"', '#', '$', '%', '&', ''', */
255, 255, 255, 255, 255, 255, 255, 255,
/* '(', ')', '*', '+', ',', '-', '.', '/', */
255, 255, 255, 62, 255, 255, 255, 63,
/* '0', '1', '2', '3', '4', '5', '6', '7', */
52, 53, 54, 55, 56, 57, 58, 59,
/* '8', '9', ':', ';', '<', '=', '>', '?', */
60, 61, 255, 255, 255, 255, 255, 255,
/* '@', 'A', 'B', 'C', 'D', 'E', 'F', 'G', */
255, 0, 1, 2, 3, 4, 5, 6,
/* 'H', 'I', 'J', 'K', 'L', 'M', 'N', 'O', */
7, 8, 9, 10, 11, 12, 13, 14,
/* 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', */
15, 16, 17, 18, 19, 20, 21, 22,
/* 'X', 'Y', 'Z', '[', '\', ']', '^', '_', */
23, 24, 25, 255, 255, 255, 255, 255,
/* '`', 'a', 'b', 'c', 'd', 'e', 'f', 'g', */
255, 26, 27, 28, 29, 30, 31, 32,
/* 'h', 'i', 'j', 'k', 'l', 'm', 'n', 'o', */
33, 34, 35, 36, 37, 38, 39, 40,
/* 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', */
41, 42, 43, 44, 45, 46, 47, 48,
/* 'x', 'y', 'z', '{', '|', '}', '~', del, */
49, 50, 51, 255, 255, 255, 255, 255
};
unsigned int
base64_encode(const unsigned char *in, unsigned int inlen, char *out)
{
int s;
unsigned int i;
unsigned int j;
unsigned char c;
unsigned char l;
s = 0;
l = 0;
for (i = j = 0; i < inlen; i++) {
c = in[i];
switch (s) {
case 0:
s = 1;
out[j++] = base64en[(c >> 2) & 0x3F];
break;
case 1:
s = 2;
out[j++] = base64en[((l & 0x3) << 4) | ((c >> 4) & 0xF)];
break;
case 2:
s = 0;
out[j++] = base64en[((l & 0xF) << 2) | ((c >> 6) & 0x3)];
out[j++] = base64en[c & 0x3F];
break;
}
l = c;
}
switch (s) {
case 1:
out[j++] = base64en[(l & 0x3) << 4];
out[j++] = BASE64_PAD;
out[j++] = BASE64_PAD;
break;
case 2:
out[j++] = base64en[(l & 0xF) << 2];
out[j++] = BASE64_PAD;
break;
}
out[j] = 0;
return j;
}
unsigned int
base64_decode(const char *in, unsigned int inlen, unsigned char *out)
{
unsigned int i;
unsigned int j;
unsigned char c;
if (inlen & 0x3) {
return 0;
}
for (i = j = 0; i < inlen; i++) {
if (in[i] == BASE64_PAD) {
break;
}
if (in[i] < BASE64DE_FIRST || in[i] > BASE64DE_LAST) {
return 0;
}
c = base64de[(unsigned char)in[i]];
if (c == 255) {
return 0;
}
switch (i & 0x3) {
case 0:
out[j] = (c << 2) & 0xFF;
break;
case 1:
out[j++] |= (c >> 4) & 0x3;
out[j] = (c & 0xF) << 4;
break;
case 2:
out[j++] |= (c >> 2) & 0xF;
out[j] = (c & 0x3) << 6;
break;
case 3:
out[j++] |= c;
break;
}
}
return j;
}
......@@ -255,66 +255,30 @@ void process_packet(void *request) {
// send the msg until ack is received
uint8_t received_ack = 0;
uint8_t pend_send_retries = PEND_SEND_RETRIES_MAX;
struct timeval tv;
tv.tv_sec = 0;
tv.tv_usec = 300000; // 300ms ack recv timeout
do {
packet_encode(
req->gch.app_key,
req->gch.dev_id,
GATEWAY_PROTOCOL_PACKET_TYPE_PEND_SEND,
payload_length, payload,
&(req->packet_length), req->packet);
do {
send_gcom_ch(&(req->gch), req->packet, req->packet_length);
// set timeout
if (setsockopt(req->gch.server_desc, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) < 0) {
perror("setsockopt error");
}
if (recv_gcom_ch(&(req->gch),
req->packet,
&(req->packet_length),
DEVICE_DATA_MAX_LENGTH) > 9)
{ /* min packet size. timeout -> -1 */
uint8_t recv_app_key[GATEWAY_PROTOCOL_APP_KEY_SIZE +1];
uint8_t recv_dev_id = 0xFF;
if (packet_decode(
recv_app_key,
&recv_dev_id,
&(req->packet_type),
&(req->packet_length), req->packet,
req->packet_length, req->packet))
{
if (!memcmp(recv_app_key, req->gch.app_key, GATEWAY_PROTOCOL_APP_KEY_SIZE) &&
recv_dev_id == req->gch.dev_id &&
req->packet_type == GATEWAY_PROTOCOL_PACKET_TYPE_STAT &&
req->packet_length == 1 &&
req->packet[0] == GATEWAY_PROTOCOL_STAT_ACK)
{
snprintf(db_query, sizeof(db_query),
"UPDATE pend_msgs SET ack = True WHERE app_key = '%s' AND dev_id = %d AND msg = '%s'",
(char *)req->gch.app_key, req->gch.dev_id, msg_cont
);
// 300 ms
usleep(300000);
pthread_mutex_lock(&mutex);
res = PQexec(conn, db_query);
pthread_mutex_unlock(&mutex);
if (PQresultStatus(res) != PGRES_COMMAND_OK) {
fprintf(stderr, "error db deleting : %s", PQerrorMessage(conn));
}
PQclear(res);
if (PQresultStatus(res) == PGRES_TUPLES_OK) {
if (!PQntuples(res) || strcmp(PQgetvalue(res, 0, 2), msg_cont)) {
received_ack = 1;
printf("ACK received\n");
} else {
printf("error: packet_type = %02X, not STAT\n");
}
}
}
PQclear(res);
printf("received_ack = %d, retries = %d\n", received_ack, pend_send_retries);
} while (!received_ack && pend_send_retries--);
// cancel timeout
tv.tv_usec = 0;
if (setsockopt(req->gch.server_desc, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) < 0) {
perror("setsockopt error");
}
} else {
gateway_protocol_mk_stat(
&(req->gch),
......@@ -325,6 +289,34 @@ void process_packet(void *request) {
printf("nothing for app %s dev %d\n", (char *)req->gch.app_key, req->gch.dev_id);
}
} else if (req->packet_type == GATEWAY_PROTOCOL_PACKET_TYPE_STAT) {
// TODO change to ACK_PEND = 0x01
if (payload[0] == 0x00) {
char db_query[200];
snprintf(db_query, sizeof(db_query),
"SELECT * FROM pend_msgs WHERE app_key = '%s' AND dev_id = %d AND ack = False",
(char *)req->gch.app_key, req->gch.dev_id
);
pthread_mutex_lock(&mutex);
res = PQexec(conn, db_query);
pthread_mutex_unlock(&mutex);
if (PQresultStatus(res) == PGRES_TUPLES_OK && PQntuples(res)) {
snprintf(db_query, sizeof(db_query),
"UPDATE pend_msgs SET ack = True WHERE app_key = '%s' AND dev_id = %d AND msg = '%s'",
(char *)req->gch.app_key, req->gch.dev_id, PQgetvalue(res, 0, 2)
);
PQclear(res);
pthread_mutex_lock(&mutex);
res = PQexec(conn, db_query);
pthread_mutex_unlock(&mutex);
if (PQresultStatus(res) == PGRES_COMMAND_OK) {
printf("pend_msgs updated\n");
} else {
fprintf(stderr, "database error : %s\n", PQerrorMessage(conn));
}
}
PQclear(res);
}
} else {
gateway_protocol_mk_stat(
&(req->gch),
......
#include <gateway_protocol.h>
#define GATEWAY_PROTOCOL_APP_KEY_SIZE 8
static uint8_t app_key[GATEWAY_PROTOCOL_APP_KEY_SIZE];
static uint8_t dev_id = 0xFF;
void gateway_protocol_init(const uint8_t *appkey, const uint8_t devid) {
memcpy(app_key, appkey, GATEWAY_PROTOCOL_APP_KEY_SIZE);
dev_id = devid;
}
void gateway_protocol_packet_encode (
const gateway_protocol_packet_type_t packet_type,
const uint8_t payload_length,
const uint8_t *payload,
uint8_t *packet_length,
uint8_t *packet)
{
*packet_length = 0;
memcpy(&packet[*packet_length], app_key, GATEWAY_PROTOCOL_APP_KEY_SIZE);
(*packet_length) += GATEWAY_PROTOCOL_APP_KEY_SIZE;
packet[*packet_length] = dev_id;
(*packet_length)++;
packet[*packet_length] = packet_type;
(*packet_length)++;
packet[*packet_length] = payload_length;
(*packet_length)++;
memcpy(&packet[*packet_length], payload, payload_length);
(*packet_length) += payload_length;
}
uint8_t gateway_protocol_packet_decode (
gateway_protocol_packet_type_t *packet_type,
uint8_t *payload_length,
uint8_t *payload,
const uint8_t packet_length,
const uint8_t *packet)
{
uint8_t p_len = 0;
uint8_t appkey[GATEWAY_PROTOCOL_APP_KEY_SIZE];
uint8_t dev;
memcpy(appkey, &packet[p_len], GATEWAY_PROTOCOL_APP_KEY_SIZE);
p_len += GATEWAY_PROTOCOL_APP_KEY_SIZE;
dev = packet[p_len];
p_len++;
*packet_type = (gateway_protocol_packet_type_t) packet[p_len];
p_len++;
p_len++;
*payload_length = packet_length - p_len;
memcpy(payload, &packet[p_len], *payload_length);
p_len += *payload_length;
return (memcmp(appkey, app_key, GATEWAY_PROTOCOL_APP_KEY_SIZE) &&
dev == dev_id &&
p_len == packet_length);
}
#include<stdio.h>
#include<string.h>
#include<stdlib.h>
#include<sys/socket.h>
#include<arpa/inet.h> //inet_addr
#include<unistd.h>
#include<stdint.h>
#include<pthread.h>
#include<gateway_protocol.h>
#include<sys/time.h>
void * connection_handler (void *args);
int main (int argc, char **argv) {
int server_desc, client_desc;
struct sockaddr_in server, client;
socklen_t client_socklen;
char buf[1024] = "";
if ((server_desc = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
perror("socket creation error");
return EXIT_FAILURE;
}
server.sin_family = AF_INET;
server.sin_port = htons(9043);
server.sin_addr.s_addr = INADDR_ANY;
if (bind(server_desc, (struct sockaddr *) &server, sizeof(server)) < 0) {
perror("binding error");
return EXIT_FAILURE;
}
listen(server_desc, 3);
printf("liteninig...\n");
while ((client_desc = accept(server_desc, (struct sockaddr *) &client, &client_socklen))) {
//strncpy(buf, "Hello, client!", sizeof(buf));
//write(client_desc, buf, strlen(buf));
printf("packet received!\n");
pthread_t thr;
if (pthread_create(&thr, NULL, connection_handler, (void *) &client_desc) < 0) {
perror("thread creation failed");
return EXIT_FAILURE;
}
pthread_join(thr, NULL);
}
return EXIT_SUCCESS;
}
void *connection_handler(void *args) {
int client_desc = *(int *)args;
uint8_t buf[128] = "";
uint8_t buf_len = 0;
uint8_t payload[128];
uint8_t payload_length = 0;
//strncpy(buf, "connection handler greetings!", sizeof(buf));
//write(client_desc, buf, strlen(buf));
if ((buf_len = recv(client_desc, buf, sizeof(buf), 0)) > 0) {
uint8_t dev_id = 0xFF;
gateway_protocol_packet_type_t packet_type;
if (gateway_protocol_packet_decode(
&dev_id,
&packet_type,
&payload_length, payload,
buf_len, buf))
{
if (packet_type == GATEWAY_PROTOCOL_PACKET_TYPE_TIME_REQ) {
printf("TIME REQ received\n");
struct timeval tv;
buf_len = 0;
buf[0] = dev_id;
buf_len++;
buf[1] = GATEWAY_PROTOCOL_PACKET_TYPE_TIME_SEND;
buf_len++;
gettimeofday(&tv, NULL);
memcpy(&buf[buf_len], &tv.tv_sec, sizeof(uint32_t));
buf_len += sizeof(uint32_t);
write(client_desc, buf, buf_len);
} else {
perror("packet type error");
}
} else {
perror("packet decode error");
}
}
}
/* vim: set et ts=3 sw=3 sts=3 ft=c:
*
* Copyright (C) 2012, 2013, 2014 James McLaughlin et al. All rights reserved.
* https://github.com/udp/json-parser
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
* 1. Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
* FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
* OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
* HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
* LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
* OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
* SUCH DAMAGE.
*/
#include "json.h"
#ifdef _MSC_VER
#ifndef _CRT_SECURE_NO_WARNINGS
#define _CRT_SECURE_NO_WARNINGS
#endif
#include <stdint.h>
#endif
const struct _json_value json_value_none;
#include <stdio.h>
#include <string.h>
#include <ctype.h>
#include <math.h>
typedef unsigned int json_uchar;
/* There has to be a better way to do this */
static const json_int_t JSON_INT_MAX = sizeof(json_int_t) == 1
? INT8_MAX
: (sizeof(json_int_t) == 2
? INT16_MAX
: (sizeof(json_int_t) == 4
? INT32_MAX
: INT64_MAX));
static unsigned char hex_value (json_char c)
{
if (isdigit(c))
return c - '0';
switch (c) {
case 'a': case 'A': return 0x0A;
case 'b': case 'B': return 0x0B;
case 'c': case 'C': return 0x0C;
case 'd': case 'D': return 0x0D;
case 'e': case 'E': return 0x0E;
case 'f': case 'F': return 0x0F;
default: return 0xFF;
}
}
static int would_overflow (json_int_t value, json_char b)
{
return ((JSON_INT_MAX - (b - '0')) / 10 ) < value;
}
typedef struct
{
unsigned long used_memory;
unsigned int uint_max;
unsigned long ulong_max;
json_settings settings;
int first_pass;
const json_char * ptr;
unsigned int cur_line, cur_col;
} json_state;
static void * default_alloc (size_t size, int zero, void * user_data)
{
return zero ? calloc (1, size) : malloc (size);
}
static void default_free (void * ptr, void * user_data)
{
free (ptr);
}
static void * json_alloc (json_state * state, unsigned long size, int zero)
{
if ((state->ulong_max - state->used_memory) < size)
return 0;
if (state->settings.max_memory
&& (state->used_memory += size) > state->settings.max_memory)
{
return 0;
}
return state->settings.mem_alloc (size, zero, state->settings.user_data);
}
static int new_value (json_state * state,
json_value ** top, json_value ** root, json_value ** alloc,
json_type type)
{
json_value * value;
int values_size;
if (!state->first_pass)
{
value = *top = *alloc;
*alloc = (*alloc)->_reserved.next_alloc;
if (!*root)
*root = value;
switch (value->type)
{
case json_array:
if (value->u.array.length == 0)
break;
if (! (value->u.array.values = (json_value **) json_alloc
(state, value->u.array.length * sizeof (json_value *), 0)) )
{
return 0;
}
value->u.array.length = 0;
break;
case json_object:
if (value->u.object.length == 0)
break;
values_size = sizeof (*value->u.object.values) * value->u.object.length;
if (! (value->u.object.values = (json_object_entry *) json_alloc
(state, values_size + ((unsigned long) value->u.object.values), 0)) )
{
return 0;
}
value->_reserved.object_mem = (*(char **) &value->u.object.values) + values_size;
value->u.object.length = 0;
break;
case json_string:
if (! (value->u.string.ptr = (json_char *) json_alloc
(state, (value->u.string.length + 1) * sizeof (json_char), 0)) )
{
return 0;
}
value->u.string.length = 0;
break;
default:
break;
};
return 1;
}
if (! (value = (json_value *) json_alloc
(state, sizeof (json_value) + state->settings.value_extra, 1)))
{
return 0;
}
if (!*root)
*root = value;
value->type = type;
value->parent = *top;
#ifdef JSON_TRACK_SOURCE
value->line = state->cur_line;
value->col = state->cur_col;
#endif
if (*alloc)
(*alloc)->_reserved.next_alloc = value;
*alloc = *top = value;
return 1;
}
#define whitespace \
case '\n': ++ state.cur_line; state.cur_col = 0; \
case ' ': case '\t': case '\r'
#define string_add(b) \
do { if (!state.first_pass) string [string_length] = b; ++ string_length; } while (0);
#define line_and_col \
state.cur_line, state.cur_col
static const long
flag_next = 1 << 0,
flag_reproc = 1 << 1,
flag_need_comma = 1 << 2,
flag_seek_value = 1 << 3,
flag_escaped = 1 << 4,
flag_string = 1 << 5,
flag_need_colon = 1 << 6,
flag_done = 1 << 7,
flag_num_negative = 1 << 8,
flag_num_zero = 1 << 9,
flag_num_e = 1 << 10,
flag_num_e_got_sign = 1 << 11,
flag_num_e_negative = 1 << 12,
flag_line_comment = 1 << 13,
flag_block_comment = 1 << 14,
flag_num_got_decimal = 1 << 15;
json_value * json_parse_ex (json_settings * settings,
const json_char * json,
size_t length,
char * error_buf)
{
json_char error [json_error_max];
const json_char * end;
json_value * top, * root, * alloc = 0;
json_state state = { 0 };
long flags = 0;
double num_digits = 0, num_e = 0;
double num_fraction = 0;
/* Skip UTF-8 BOM
*/
if (length >= 3 && ((unsigned char) json [0]) == 0xEF
&& ((unsigned char) json [1]) == 0xBB
&& ((unsigned char) json [2]) == 0xBF)
{
json += 3;
length -= 3;
}
error[0] = '\0';
end = (json + length);
memcpy (&state.settings, settings, sizeof (json_settings));
if (!state.settings.mem_alloc)
state.settings.mem_alloc = default_alloc;
if (!state.settings.mem_free)
state.settings.mem_free = default_free;
memset (&state.uint_max, 0xFF, sizeof (state.uint_max));
memset (&state.ulong_max, 0xFF, sizeof (state.ulong_max));
state.uint_max -= 8; /* limit of how much can be added before next check */
state.ulong_max -= 8;
for (state.first_pass = 1; state.first_pass >= 0; -- state.first_pass)
{
json_uchar uchar;
unsigned char uc_b1, uc_b2, uc_b3, uc_b4;
json_char * string = 0;
unsigned int string_length = 0;
top = root = 0;
flags = flag_seek_value;
state.cur_line = 1;
for (state.ptr = json ;; ++ state.ptr)
{
json_char b = (state.ptr == end ? 0 : *state.ptr);
if (flags & flag_string)
{
if (!b)
{ sprintf (error, "Unexpected EOF in string (at %d:%d)", line_and_col);
goto e_failed;
}
if (string_length > state.uint_max)
goto e_overflow;
if (flags & flag_escaped)
{
flags &= ~ flag_escaped;
switch (b)
{
case 'b': string_add ('\b'); break;
case 'f': string_add ('\f'); break;
case 'n': string_add ('\n'); break;
case 'r': string_add ('\r'); break;
case 't': string_add ('\t'); break;
case 'u':
if (end - state.ptr <= 4 ||
(uc_b1 = hex_value (*++ state.ptr)) == 0xFF ||
(uc_b2 = hex_value (*++ state.ptr)) == 0xFF ||
(uc_b3 = hex_value (*++ state.ptr)) == 0xFF ||
(uc_b4 = hex_value (*++ state.ptr)) == 0xFF)
{
sprintf (error, "Invalid character value `%c` (at %d:%d)", b, line_and_col);
goto e_failed;
}
uc_b1 = (uc_b1 << 4) | uc_b2;
uc_b2 = (uc_b3 << 4) | uc_b4;
uchar = (uc_b1 << 8) | uc_b2;
if ((uchar & 0xF800) == 0xD800) {
json_uchar uchar2;
if (end - state.ptr <= 6 || (*++ state.ptr) != '\\' || (*++ state.ptr) != 'u' ||
(uc_b1 = hex_value (*++ state.ptr)) == 0xFF ||
(uc_b2 = hex_value (*++ state.ptr)) == 0xFF ||
(uc_b3 = hex_value (*++ state.ptr)) == 0xFF ||
(uc_b4 = hex_value (*++ state.ptr)) == 0xFF)
{
sprintf (error, "Invalid character value `%c` (at %d:%d)", b, line_and_col);
goto e_failed;
}
uc_b1 = (uc_b1 << 4) | uc_b2;
uc_b2 = (uc_b3 << 4) | uc_b4;
uchar2 = (uc_b1 << 8) | uc_b2;
uchar = 0x010000 | ((uchar & 0x3FF) << 10) | (uchar2 & 0x3FF);
}
if (sizeof (json_char) >= sizeof (json_uchar) || (uchar <= 0x7F))
{
string_add ((json_char) uchar);
break;
}
if (uchar <= 0x7FF)
{
if (state.first_pass)
string_length += 2;
else
{ string [string_length ++] = 0xC0 | (uchar >> 6);
string [string_length ++] = 0x80 | (uchar & 0x3F);
}
break;
}
if (uchar <= 0xFFFF) {
if (state.first_pass)
string_length += 3;
else
{ string [string_length ++] = 0xE0 | (uchar >> 12);
string [string_length ++] = 0x80 | ((uchar >> 6) & 0x3F);
string [string_length ++] = 0x80 | (uchar & 0x3F);
}
break;
}
if (state.first_pass)
string_length += 4;
else
{ string [string_length ++] = 0xF0 | (uchar >> 18);
string [string_length ++] = 0x80 | ((uchar >> 12) & 0x3F);
string [string_length ++] = 0x80 | ((uchar >> 6) & 0x3F);
string [string_length ++] = 0x80 | (uchar & 0x3F);
}
break;
default:
string_add (b);
};
continue;
}
if (b == '\\')
{
flags |= flag_escaped;
continue;
}
if (b == '"')
{
if (!state.first_pass)
string [string_length] = 0;
flags &= ~ flag_string;
string = 0;
switch (top->type)
{
case json_string:
top->u.string.length = string_length;
flags |= flag_next;
break;
case json_object:
if (state.first_pass)
(*(json_char **) &top->u.object.values) += string_length + 1;
else
{
top->u.object.values [top->u.object.length].name
= (json_char *) top->_reserved.object_mem;
top->u.object.values [top->u.object.length].name_length
= string_length;
(*(json_char **) &top->_reserved.object_mem) += string_length + 1;
}
flags |= flag_seek_value | flag_need_colon;
continue;
default:
break;
};
}
else
{
string_add (b);
continue;
}
}
if (state.settings.settings & json_enable_comments)
{
if (flags & (flag_line_comment | flag_block_comment))
{
if (flags & flag_line_comment)
{
if (b == '\r' || b == '\n' || !b)
{
flags &= ~ flag_line_comment;
-- state.ptr; /* so null can be reproc'd */
}
continue;
}
if (flags & flag_block_comment)
{
if (!b)
{ sprintf (error, "%d:%d: Unexpected EOF in block comment", line_and_col);
goto e_failed;
}
if (b == '*' && state.ptr < (end - 1) && state.ptr [1] == '/')
{
flags &= ~ flag_block_comment;
++ state.ptr; /* skip closing sequence */
}
continue;
}
}
else if (b == '/')
{
if (! (flags & (flag_seek_value | flag_done)) && top->type != json_object)
{ sprintf (error, "%d:%d: Comment not allowed here", line_and_col);
goto e_failed;
}
if (++ state.ptr == end)
{ sprintf (error, "%d:%d: EOF unexpected", line_and_col);
goto e_failed;
}
switch (b = *state.ptr)
{
case '/':
flags |= flag_line_comment;
continue;
case '*':
flags |= flag_block_comment;
continue;
default:
sprintf (error, "%d:%d: Unexpected `%c` in comment opening sequence", line_and_col, b);
goto e_failed;
};
}
}
if (flags & flag_done)
{
if (!b)
break;
switch (b)
{
whitespace:
continue;
default:
sprintf (error, "%d:%d: Trailing garbage: `%c`",
state.cur_line, state.cur_col, b);
goto e_failed;
};
}
if (flags & flag_seek_value)
{
switch (b)
{
whitespace:
continue;
case ']':
if (top && top->type == json_array)
flags = (flags & ~ (flag_need_comma | flag_seek_value)) | flag_next;
else
{ sprintf (error, "%d:%d: Unexpected ]", line_and_col);
goto e_failed;
}
break;
default:
if (flags & flag_need_comma)
{
if (b == ',')
{ flags &= ~ flag_need_comma;
continue;
}
else
{
sprintf (error, "%d:%d: Expected , before %c",
state.cur_line, state.cur_col, b);
goto e_failed;
}
}
if (flags & flag_need_colon)
{
if (b == ':')
{ flags &= ~ flag_need_colon;
continue;
}
else
{
sprintf (error, "%d:%d: Expected : before %c",
state.cur_line, state.cur_col, b);
goto e_failed;
}
}
flags &= ~ flag_seek_value;
switch (b)
{
case '{':
if (!new_value (&state, &top, &root, &alloc, json_object))
goto e_alloc_failure;
continue;
case '[':
if (!new_value (&state, &top, &root, &alloc, json_array))
goto e_alloc_failure;
flags |= flag_seek_value;
continue;
case '"':
if (!new_value (&state, &top, &root, &alloc, json_string))
goto e_alloc_failure;
flags |= flag_string;
string = top->u.string.ptr;
string_length = 0;
continue;
case 't':
if ((end - state.ptr) < 3 || *(++ state.ptr) != 'r' ||
*(++ state.ptr) != 'u' || *(++ state.ptr) != 'e')
{
goto e_unknown_value;
}
if (!new_value (&state, &top, &root, &alloc, json_boolean))
goto e_alloc_failure;
top->u.boolean = 1;
flags |= flag_next;
break;
case 'f':
if ((end - state.ptr) < 4 || *(++ state.ptr) != 'a' ||
*(++ state.ptr) != 'l' || *(++ state.ptr) != 's' ||
*(++ state.ptr) != 'e')
{
goto e_unknown_value;
}
if (!new_value (&state, &top, &root, &alloc, json_boolean))
goto e_alloc_failure;
flags |= flag_next;
break;
case 'n':
if ((end - state.ptr) < 3 || *(++ state.ptr) != 'u' ||
*(++ state.ptr) != 'l' || *(++ state.ptr) != 'l')
{
goto e_unknown_value;
}
if (!new_value (&state, &top, &root, &alloc, json_null))
goto e_alloc_failure;
flags |= flag_next;
break;
default:
if (isdigit (b) || b == '-')
{
if (!new_value (&state, &top, &root, &alloc, json_integer))
goto e_alloc_failure;
if (!state.first_pass)
{
while (isdigit (b) || b == '+' || b == '-'
|| b == 'e' || b == 'E' || b == '.')
{
if ( (++ state.ptr) == end)
{
b = 0;
break;
}
b = *state.ptr;
}
flags |= flag_next | flag_reproc;
break;
}
flags &= ~ (flag_num_negative | flag_num_e |
flag_num_e_got_sign | flag_num_e_negative |
flag_num_zero);
num_digits = 0;
num_fraction = 0;
num_e = 0;
if (b != '-')
{
flags |= flag_reproc;
break;
}
flags |= flag_num_negative;
continue;
}
else
{ sprintf (error, "%d:%d: Unexpected %c when seeking value", line_and_col, b);
goto e_failed;
}
};
};
}
else
{
switch (top->type)
{
case json_object:
switch (b)
{
whitespace:
continue;
case '"':
if (flags & flag_need_comma)
{ sprintf (error, "%d:%d: Expected , before \"", line_and_col);
goto e_failed;
}
flags |= flag_string;
string = (json_char *) top->_reserved.object_mem;
string_length = 0;
break;
case '}':
flags = (flags & ~ flag_need_comma) | flag_next;
break;
case ',':
if (flags & flag_need_comma)
{
flags &= ~ flag_need_comma;
break;
}
default:
sprintf (error, "%d:%d: Unexpected `%c` in object", line_and_col, b);
goto e_failed;
};
break;
case json_integer:
case json_double:
if (isdigit (b))
{
++ num_digits;
if (top->type == json_integer || flags & flag_num_e)
{
if (! (flags & flag_num_e))
{
if (flags & flag_num_zero)
{ sprintf (error, "%d:%d: Unexpected `0` before `%c`", line_and_col, b);
goto e_failed;
}
if (num_digits == 1 && b == '0')
flags |= flag_num_zero;
}
else
{
flags |= flag_num_e_got_sign;
num_e = (num_e * 10) + (b - '0');
continue;
}
if (would_overflow(top->u.integer, b))
{ -- num_digits;
-- state.ptr;
top->type = json_double;
top->u.dbl = (double)top->u.integer;
continue;
}
top->u.integer = (top->u.integer * 10) + (b - '0');
continue;
}
if (flags & flag_num_got_decimal)
num_fraction = (num_fraction * 10) + (b - '0');
else
top->u.dbl = (top->u.dbl * 10) + (b - '0');
continue;
}
if (b == '+' || b == '-')
{
if ( (flags & flag_num_e) && !(flags & flag_num_e_got_sign))
{
flags |= flag_num_e_got_sign;
if (b == '-')
flags |= flag_num_e_negative;
continue;
}
}
else if (b == '.' && top->type == json_integer)
{
if (!num_digits)
{ sprintf (error, "%d:%d: Expected digit before `.`", line_and_col);
goto e_failed;
}
top->type = json_double;
top->u.dbl = (double) top->u.integer;
flags |= flag_num_got_decimal;
num_digits = 0;
continue;
}
if (! (flags & flag_num_e))
{
if (top->type == json_double)
{
if (!num_digits)
{ sprintf (error, "%d:%d: Expected digit after `.`", line_and_col);
goto e_failed;
}
top->u.dbl += num_fraction / pow (10.0, num_digits);
}
if (b == 'e' || b == 'E')
{
flags |= flag_num_e;
if (top->type == json_integer)
{
top->type = json_double;
top->u.dbl = (double) top->u.integer;
}
num_digits = 0;
flags &= ~ flag_num_zero;
continue;
}
}
else
{
if (!num_digits)
{ sprintf (error, "%d:%d: Expected digit after `e`", line_and_col);
goto e_failed;
}
top->u.dbl *= pow (10.0, (flags & flag_num_e_negative ? - num_e : num_e));
}
if (flags & flag_num_negative)
{
if (top->type == json_integer)
top->u.integer = - top->u.integer;
else
top->u.dbl = - top->u.dbl;
}
flags |= flag_next | flag_reproc;
break;
default:
break;
};
}
if (flags & flag_reproc)
{
flags &= ~ flag_reproc;
-- state.ptr;
}
if (flags & flag_next)
{
flags = (flags & ~ flag_next) | flag_need_comma;
if (!top->parent)
{
/* root value done */
flags |= flag_done;
continue;
}
if (top->parent->type == json_array)
flags |= flag_seek_value;
if (!state.first_pass)
{
json_value * parent = top->parent;
switch (parent->type)
{
case json_object:
parent->u.object.values
[parent->u.object.length].value = top;
break;
case json_array:
parent->u.array.values
[parent->u.array.length] = top;
break;
default:
break;
};
}
if ( (++ top->parent->u.array.length) > state.uint_max)
goto e_overflow;
top = top->parent;
continue;
}
}
alloc = root;
}
return root;
e_unknown_value:
sprintf (error, "%d:%d: Unknown value", line_and_col);
goto e_failed;
e_alloc_failure:
strcpy (error, "Memory allocation failure");
goto e_failed;
e_overflow:
sprintf (error, "%d:%d: Too long (caught overflow)", line_and_col);
goto e_failed;
e_failed:
if (error_buf)
{
if (*error)
strcpy (error_buf, error);
else
strcpy (error_buf, "Unknown error");
}
if (state.first_pass)
alloc = root;
while (alloc)
{
top = alloc->_reserved.next_alloc;
state.settings.mem_free (alloc, state.settings.user_data);
alloc = top;
}
if (!state.first_pass)
json_value_free_ex (&state.settings, root);
return 0;
}
json_value * json_parse (const json_char * json, size_t length)
{
json_settings settings = { 0 };
return json_parse_ex (&settings, json, length, 0);
}
void json_value_free_ex (json_settings * settings, json_value * value)
{
json_value * cur_value;
if (!value)
return;
value->parent = 0;
while (value)
{
switch (value->type)
{
case json_array:
if (!value->u.array.length)
{
settings->mem_free (value->u.array.values, settings->user_data);
break;
}
value = value->u.array.values [-- value->u.array.length];
continue;
case json_object:
if (!value->u.object.length)
{
settings->mem_free (value->u.object.values, settings->user_data);
break;
}
value = value->u.object.values [-- value->u.object.length].value;
continue;
case json_string:
settings->mem_free (value->u.string.ptr, settings->user_data);
break;
default:
break;
};
cur_value = value;
value = value->parent;
settings->mem_free (cur_value, settings->user_data);
}
}
void json_value_free (json_value * value)
{
json_settings settings = { 0 };
settings.mem_free = default_free;
json_value_free_ex (&settings, value);
}
#include <stdlib.h>
#include <pthread.h>
#include "task_queue.h"
struct queue_job {
task_func_t func;
void *arg;
struct queue_job *next;
};
typedef struct queue_job queue_job_t;
struct task_queue {
queue_job_t *next;
queue_job_t *last;
pthread_mutex_t mutex;
int active_tasks;
int size;
int suspended;
};
static int max_active_tasks;
static queue_job_t * queue_job_create(task_func_t func, void *arg);
static void queue_job_destroy(queue_job_t *qj);
static queue_job_t * queue_job_get_next(task_queue_t *tq);
static void * queue_job_exec(void *arg_tq);
task_queue_t * task_queue_create(const int max_threads) {
task_queue_t *tq = (task_queue_t *)malloc(sizeof(task_queue_t));
if (!tq) {
return NULL;
}
tq->next = NULL;
tq->last = NULL;
tq->active_tasks = 0;
tq->size = 0;
tq->suspended = 0;
pthread_mutex_init(&(tq->mutex), NULL);
max_active_tasks = max_threads > 0 ? max_threads : 1;
return tq;
}
void task_queue_destroy(task_queue_t *tq) {
queue_job_t *qj1, *qj2;
if (!tq) {
return;
}
pthread_mutex_lock(&(tq->mutex));
tq->suspended = 1;
qj1 = tq->next;
while (qj1) {
qj2 = qj1->next;
queue_job_destroy(qj1);
qj1 = qj2;
}
pthread_mutex_destroy(&(tq->mutex));
free(tq);
}
int task_queue_enqueue(task_queue_t *tq, task_func_t task, void *arg) {
queue_job_t *qj;
pthread_t thread;
int i, can_run, need_run, run, ret;
if (!tq || !task) {
return -1;
}
qj = queue_job_create(task, arg);
if (!qj) {
return -1;
}
pthread_mutex_lock(&(tq->mutex));
if (!tq->next) {
tq->next = qj;
tq->last = qj;
} else {
tq->last->next = qj; // assign next
tq->last = qj; // move pointer
}
tq->size++;
if ((tq->active_tasks < max_active_tasks) & !tq->suspended) {
can_run = max_active_tasks - tq->active_tasks;
need_run = tq->size - tq->active_tasks;
run = need_run < can_run ? need_run : can_run;
for (i = 0; i < run; i++) {
if (pthread_create(&thread, NULL, queue_job_exec, tq) < 0) {
// perror("new thread creation error");
} else {
pthread_detach(thread);
tq->active_tasks++;
}
}
}
ret = tq->active_tasks;
pthread_mutex_unlock(&(tq->mutex));
return ret;
}
void task_queue_suspend(task_queue_t *tq) {
if (!tq) {
return;
}
pthread_mutex_lock(&(tq->mutex));
tq->suspended = 1;
pthread_mutex_unlock(&(tq->mutex));
}
void task_queue_unsuspend(task_queue_t *tq) {
pthread_t thread;
int i, can_run, need_run, run;
if (!tq) {
return;
}
pthread_mutex_lock(&(tq->mutex));
tq->suspended = 0;
if (tq->size && tq->active_tasks < max_active_tasks) {
can_run = max_active_tasks - tq->active_tasks;
need_run = tq->size - tq->active_tasks;
run = need_run < can_run ? need_run : can_run;
for (i = 0; i < run; i++) {
if (pthread_create(&thread, NULL, &queue_job_exec, tq) < 0) {
// perror("new thread creation error");
} else {
pthread_detach(thread);
tq->active_tasks++;
}
}
}
pthread_mutex_unlock(&(tq->mutex));
}
int task_queue_get_size(task_queue_t *tq) {
int res;
pthread_mutex_lock(&(tq->mutex));
res = tq->size;
pthread_mutex_unlock(&(tq->mutex));
return res;
}
int task_queue_is_empty(task_queue_t *tq) {
return task_queue_get_size(tq) == 0;
}
static queue_job_t * queue_job_create(task_func_t func, void *arg) {
queue_job_t *qj;
if (!func) {
return NULL;
}
qj = (queue_job_t *)malloc(sizeof(queue_job_t));
qj->func = func;
qj->arg = arg;
qj->next = NULL;
return qj;
}
static void queue_job_destroy(queue_job_t *qj) {
if (!qj) {
return;
}
free(qj);
}
static queue_job_t * queue_job_get_next(task_queue_t *tq) {
queue_job_t *qj;
if (!tq) {
return NULL;
}
pthread_mutex_lock(&(tq->mutex));
if (tq->next) {
qj = tq->next;
tq->next = qj->next;
} else {
qj = NULL;
}
pthread_mutex_unlock(&(tq->mutex));
return qj;
}
static void * queue_job_exec(void *arg_tq) {
task_queue_t *tq = (task_queue_t *)arg_tq;
queue_job_t *qj;
pthread_t thread;
int i, can_run, need_run, run;
if (!tq) {
return NULL;
}
qj = queue_job_get_next(tq);
if (qj && qj->func) {
qj->func(qj->arg);
}
queue_job_destroy(qj);
pthread_mutex_lock(&(tq->mutex));
tq->active_tasks--;
tq->size--;
if (tq->size && tq->active_tasks < max_active_tasks && !tq->suspended) {
can_run = max_active_tasks - tq->active_tasks;
need_run = tq->size - tq->active_tasks;
run = need_run < can_run ? need_run : can_run;
for (i = 0; i < run; i++) {
if (pthread_create(&thread, NULL, &queue_job_exec, tq) < 0) {
// perror("new thread creation error");
} else {
pthread_detach(thread);
tq->active_tasks++;
}
}
}
pthread_mutex_unlock(&(tq->mutex));
return NULL;
}
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment