Libkvstorage client is now deprecated as I plan to replace it with hiredis library.
# Librspamdclient
-SET(LIBRSPAMDSRC client/librspamdclient.c ../src/mem_pool.c ../src/upstream.c ../src/printf.c)
+SET(LIBRSPAMDSRC client/librspamdclient.c
+ ../src/mem_pool.c
+ ../src/upstream.c
+ ../src/printf.c)
ADD_LIBRARY(rspamdclient SHARED ${LIBRSPAMDSRC})
ADD_LIBRARY(rspamdclient_static STATIC ${LIBRSPAMDSRC})
# Libkvstorageclient
-SET(LIBRKVSTORAGESRC kvstorage/libkvstorageclient.c ../src/mem_pool.c ../src/upstream.c ../src/printf.c ../src/util.c)
+SET(LIBRKVSTORAGESRC kvstorage/libkvstorageclient.c
+ ../src/mem_pool.c
+ ../src/upstream.c
+ ../src/printf.c
+ ../src/util.c)
ADD_LIBRARY(kvstorageclient SHARED ${LIBRKVSTORAGESRC})
ADD_LIBRARY(kvstorageclient_static STATIC ${LIBRKVSTORAGESRC})
TARGET_LINK_LIBRARIES(kvstorageclient ${CMAKE_REQUIRED_LIBRARIES})
TARGET_LINK_LIBRARIES(kvstorageclient pcre)
TARGET_LINK_LIBRARIES(kvstorageclient ${GLIB2_LIBRARIES})
+TARGET_LINK_LIBRARIES(kvstorageclient event)
TARGET_LINK_LIBRARIES(kvstorageclient_static ${CMAKE_REQUIRED_LIBRARIES})
TARGET_LINK_LIBRARIES(kvstorageclient_static ${GLIB2_LIBRARIES})
#include "config.h"
-#include "main.h"
+#include "mem_pool.h"
+#include "util.h"
#include "libkvstorageclient.h"
#define MAX_KV_LINE 1024
+#ifdef CRLF
+#undef CRLF
+#undef CR
+#undef LF
+#endif
+
+#define CRLF "\r\n"
+#define CR '\r'
+#define LF '\n'
+
struct kvstorage_buf {
guint pos;
guint len;
static enum rspamd_kvstorage_error
rspamd_kvstorage_parse_get_line (struct kvstorage_buf *buf, guint *len, guint *flags)
{
- guint8 *p, *c;
+ guint8 *p, *c, *end;
gboolean error = TRUE;
gchar *err_str;
p = buf->data;
- while (p - buf->data < buf->pos) {
+ end = buf->data + buf->pos;
+ while (p < end) {
if (g_ascii_isspace (*p)) {
error = FALSE;
while (p - buf->data < buf->pos && g_ascii_isspace (*p)) {
/* Here we got key, flags and size items */
/* Skip key */
error = TRUE;
- while (p - buf->data < buf->pos) {
+ while (p < end) {
if (g_ascii_isspace (*p)) {
error = FALSE;
- while (p - buf->data < buf->pos && g_ascii_isspace (*p)) {
+ /* Skip spaces after key */
+ while (p < end && g_ascii_isspace (*p)) {
p ++;
}
break;
/* Read flags */
c = p;
error = TRUE;
- while (p - buf->data < buf->pos) {
+ while (p < end) {
if (g_ascii_isspace (*p)) {
error = FALSE;
+ /* Skip spaces after flags */
while (p - buf->data < buf->pos && g_ascii_isspace (*p)) {
p ++;
}
}
/* Read len */
c = p;
- error = TRUE;
- while (p - buf->data < buf->pos) {
- if (g_ascii_isspace (*p)) {
- error = FALSE;
- while (p - buf->data < buf->pos && g_ascii_isspace (*p)) {
- p ++;
- }
- break;
- }
- else if (!g_ascii_isdigit (*p)) {
+ while (p < end) {
+ if (!g_ascii_isdigit (*p)) {
break;
}
p ++;
cb (KVSTORAGE_ERROR_TIMEOUT, d->c, d->ud);
}
else {
+ d->c->state = KV_STATE_CONNECTED;
cb (KVSTORAGE_ERROR_OK, d->c, d->ud);
}
- d->c->state = KV_STATE_CONNECTED;
}
static void
cb = (kvstorage_read_cb)d->c->read_cb;
if (what == EV_TIMEOUT) {
- cb (KVSTORAGE_ERROR_TIMEOUT, d->key, NULL, 0, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_TIMEOUT, d->key, d->keylen, NULL, 0, d->c, d->ud);
return;
}
if (d->c->state == KV_STATE_GET) {
}
else if (r == 0) {
/* We have written everything */
+ d->buf = rspamd_kvstorage_buf_create (MAX_KV_LINE, d->c->pool);
d->c->state = KV_STATE_READ_ELT;
event_del (&d->c->ev);
event_set (&d->c->ev, d->c->sock, EV_READ, rspamd_kvstorage_read_cb, d);
}
else {
/* Error occured during writing */
- cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, NULL, 0, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->keylen, NULL, 0, d->c, d->ud);
}
}
else if (d->c->state == KV_STATE_WRITE_DATA) {
else if (r == 0) {
/* We have written everything */
d->c->state = KV_STATE_READ_ELT;
+ d->buf = rspamd_kvstorage_buf_create (MAX_KV_LINE, d->c->pool);
event_del (&d->c->ev);
event_set (&d->c->ev, d->c->sock, EV_READ, rspamd_kvstorage_read_cb, d);
if (d->c->tv.tv_sec != 0 || d->c->tv.tv_usec != 0) {
}
else {
/* Error occured during writing */
- cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, NULL, 0, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->keylen, NULL, 0, d->c, d->ud);
}
}
else if (d->c->state == KV_STATE_READ_ELT) {
else if (r == 0) {
/* Got all data about elt */
if ((r = rspamd_kvstorage_parse_get_line (d->buf, &d->datalen, &flags)) != KVSTORAGE_ERROR_OK) {
- cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, NULL, 0, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->keylen, NULL, 0, d->c, d->ud);
return;
}
rspamd_kvstorage_buf_drainline (d->buf);
+
/* Now allocate and read the data */
databuf = rspamd_kvstorage_buf_create (d->datalen, d->c->pool);
memcpy (databuf->data, d->buf->data, d->buf->pos);
}
else {
/* Error occured during reading reply line */
- cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, NULL, 0, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->keylen, NULL, 0, d->c, d->ud);
}
}
else if (d->c->state == KV_STATE_READ_DATA) {
d->c->state = KV_STATE_READ_REPLY;
/* Save databuf */
d->data = d->buf->data;
- d->buf = rspamd_kvstorage_buf_create (MAX_KV_LINE, d->c->pool);
event_del (&d->c->ev);
event_set (&d->c->ev, d->c->sock, EV_READ, rspamd_kvstorage_read_cb, d);
if (d->c->tv.tv_sec != 0 || d->c->tv.tv_usec != 0) {
}
else {
/* Error occured */
- cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, NULL, 0, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->keylen, NULL, 0, d->c, d->ud);
}
}
else if (d->c->state == KV_STATE_READ_REPLY) {
}
}
else if (r == 0) {
- cb (rspamd_kvstorage_parse_reply_error (d->buf), d->key, d->data, d->datalen, d->c, d->ud);
+ d->c->state = KV_STATE_CONNECTED;
+ cb (rspamd_kvstorage_parse_reply_error (d->buf), d->key, d->keylen, d->data, d->datalen, d->c, d->ud);
}
else {
/* Error occured during reading reply line */
- cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, NULL, 0, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->keylen, NULL, 0, d->c, d->ud);
}
}
}
cb = (kvstorage_write_cb)d->c->write_cb;
if (what == EV_TIMEOUT) {
- cb (KVSTORAGE_ERROR_TIMEOUT, d->key, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_TIMEOUT, d->key, d->keylen, d->c, d->ud);
return;
}
if (d->c->state == KV_STATE_SET) {
}
else if (r == 0) {
/* We have written everything */
+ d->buf = rspamd_kvstorage_buf_create (MAX_KV_LINE, d->c->pool);
d->c->state = KV_STATE_READ_REPLY;
event_del (&d->c->ev);
event_set (&d->c->ev, d->c->sock, EV_READ, rspamd_kvstorage_write_cb, d);
}
else {
/* Error occured during writing */
- cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->keylen, d->c, d->ud);
}
}
else if (d->c->state == KV_STATE_WRITE_DATA) {
else if (r == 0) {
/* We have written everything */
d->c->state = KV_STATE_READ_REPLY;
+ d->buf = rspamd_kvstorage_buf_create (MAX_KV_LINE, d->c->pool);
event_del (&d->c->ev);
event_set (&d->c->ev, d->c->sock, EV_READ, rspamd_kvstorage_write_cb, d);
if (d->c->tv.tv_sec != 0 || d->c->tv.tv_usec != 0) {
}
else {
/* Error occured during writing */
- cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->keylen, d->c, d->ud);
}
}
else if (d->c->state == KV_STATE_READ_REPLY) {
}
}
else if (r == 0) {
- cb (rspamd_kvstorage_parse_reply_error (d->buf), d->key, d->c, d->ud);
+ d->c->state = KV_STATE_CONNECTED;
+ cb (rspamd_kvstorage_parse_reply_error (d->buf), d->key, d->keylen, d->c, d->ud);
}
else {
/* Error occured during reading reply line */
- cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->keylen, d->c, d->ud);
}
}
}
cb = (kvstorage_write_cb)d->c->write_cb;
if (what == EV_TIMEOUT) {
- cb (KVSTORAGE_ERROR_TIMEOUT, d->key, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_TIMEOUT, d->key, d->keylen, d->c, d->ud);
return;
}
if (d->c->state == KV_STATE_SET) {
}
else if (r == 0) {
/* We have written everything */
+ d->buf = rspamd_kvstorage_buf_create (MAX_KV_LINE, d->c->pool);
d->c->state = KV_STATE_READ_REPLY;
event_del (&d->c->ev);
event_set (&d->c->ev, d->c->sock, EV_READ, rspamd_kvstorage_delete_cb, d);
}
else {
/* Error occured during writing */
- cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->keylen, d->c, d->ud);
}
}
else if (d->c->state == KV_STATE_WRITE_DATA) {
else if (r == 0) {
/* We have written everything */
d->c->state = KV_STATE_READ_REPLY;
+ d->buf = rspamd_kvstorage_buf_create (MAX_KV_LINE, d->c->pool);
event_del (&d->c->ev);
event_set (&d->c->ev, d->c->sock, EV_READ, rspamd_kvstorage_delete_cb, d);
if (d->c->tv.tv_sec != 0 || d->c->tv.tv_usec != 0) {
}
else {
/* Error occured during writing */
- cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->keylen, d->c, d->ud);
}
}
else if (d->c->state == KV_STATE_READ_REPLY) {
}
}
else if (r == 0) {
- cb (rspamd_kvstorage_parse_reply_error (d->buf), d->key, d->c, d->ud);
+ d->c->state = KV_STATE_CONNECTED;
+ cb (rspamd_kvstorage_parse_reply_error (d->buf), d->key, d->keylen, d->c, d->ud);
}
else {
/* Error occured during reading reply line */
- cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->c, d->ud);
+ cb (KVSTORAGE_ERROR_INTERNAL_ERROR, d->key, d->keylen, d->c, d->ud);
}
}
}
gint sock;
/* Here we do NOT try to resolve hostname */
- if ((sock = make_universal_stream_socket (host, port, TRUE, FALSE, FALSE)) == -1) {
+ if ((sock = make_universal_stream_socket (host, port, TRUE, FALSE, TRUE)) == -1) {
return KVSTORAGE_ERROR_SERVER_ERROR;
}
*/
enum rspamd_kvstorage_error
rspamd_kvstorage_get_async (struct rspamd_kvstorage_connection *conn,
- const gpointer key, kvstorage_read_cb cb, gpointer ud)
+ const gpointer key, guint keylen, kvstorage_read_cb cb, gpointer ud)
{
struct rspamd_kvstorage_async_data *d;
d->c = conn;
d->ud = ud;
d->key = memory_pool_strdup (conn->pool, key);
- d->keylen = strlen (d->key);
+ d->keylen = keylen;
conn->state = KV_STATE_GET;
/* Set event */
*/
enum rspamd_kvstorage_error
rspamd_kvstorage_set_async (struct rspamd_kvstorage_connection *conn,
- const gpointer key, const gpointer value, gsize len, guint expire, kvstorage_write_cb cb, gpointer ud)
+ const gpointer key, guint keylen, const gpointer value, gsize len, guint expire, kvstorage_write_cb cb,
+ gpointer ud)
{
struct rspamd_kvstorage_async_data *d;
d->c = conn;
d->ud = ud;
d->key = memory_pool_strdup (conn->pool, key);
- d->keylen = strlen (d->key);
+ d->keylen = keylen;
d->data = value;
d->datalen = len;
conn->state = KV_STATE_SET;
*/
enum rspamd_kvstorage_error
rspamd_kvstorage_delete_async (struct rspamd_kvstorage_connection *conn,
- const gpointer key, kvstorage_write_cb cb, gpointer ud)
+ const gpointer key, guint keylen, kvstorage_write_cb cb, gpointer ud)
{
struct rspamd_kvstorage_async_data *d;
d->c = conn;
d->ud = ud;
d->key = memory_pool_strdup (conn->pool, key);
- d->keylen = strlen (d->key);
+ d->keylen = keylen;
conn->state = KV_STATE_SET;
/* Set event */
*/
enum rspamd_kvstorage_error
rspamd_kvstorage_get_sync (struct rspamd_kvstorage_connection *conn,
- const gpointer key, gpointer **value, guint *len)
+ const gpointer key, guint keylen, gpointer **value, guint *len)
{
struct kvstorage_buf *buf, *databuf;
gint r;
buf = rspamd_kvstorage_buf_create (MAX_KV_LINE, conn->pool);
- r = rspamd_snprintf (buf->data, buf->len, "get %s" CRLF, key);
+ r = rspamd_snprintf (buf->data, buf->len, "get %*s" CRLF, keylen, key);
buf->len = r;
while ((r = rspamd_kvstorage_buf_writeall (buf, conn)) > 0) {
poll_sync_socket (conn->sock, tv_to_msec (&conn->tv), POLL_OUT);
*/
enum rspamd_kvstorage_error
rspamd_kvstorage_set_sync (struct rspamd_kvstorage_connection *conn,
- const gpointer key, const gpointer value, gsize len, guint expire)
+ const gpointer key, guint keylen, const gpointer value, gsize len, guint expire)
{
struct kvstorage_buf *buf;
- gint r, keylen, buflen;
+ gint r, buflen;
if (conn == NULL || conn->state != KV_STATE_CONNECTED) {
return KVSTORAGE_ERROR_INTERNAL_ERROR;
}
/* Create buf */
- keylen = strlen (key);
buflen = len + keylen + sizeof ("set 4294967296 4294967296 4294967296" CRLF);
buf = rspamd_kvstorage_buf_create (buflen, conn->pool);
*/
enum rspamd_kvstorage_error
rspamd_kvstorage_delete_sync (struct rspamd_kvstorage_connection *conn,
- const gpointer key)
+ const gpointer key, guint keylen)
{
struct kvstorage_buf *buf;
- gint r, keylen, buflen;
+ gint r, buflen;
if (conn == NULL || conn->state != KV_STATE_CONNECTED) {
return KVSTORAGE_ERROR_INTERNAL_ERROR;
}
/* Create buf */
- keylen = strlen (key);
buflen = MAX (keylen + sizeof ("delete " CRLF), MAX_KV_LINE);
buf = rspamd_kvstorage_buf_create (buflen, conn->pool);
return KVSTORAGE_ERROR_OK;
}
+
+const gchar*
+rspamd_kvstorage_strerror (enum rspamd_kvstorage_error err)
+{
+ switch (err) {
+ case KVSTORAGE_ERROR_OK:
+ return "operation completed";
+ case KVSTORAGE_ERROR_TIMEOUT:
+ return "operation timeout";
+ case KVSTORAGE_ERROR_NOT_FOUND:
+ return "key not found";
+ case KVSTORAGE_ERROR_NOT_STORED:
+ return "key not stored";
+ case KVSTORAGE_ERROR_EXISTS:
+ return "key exists";
+ case KVSTORAGE_ERROR_SERVER_ERROR:
+ return "server error";
+ case KVSTORAGE_ERROR_CLIENT_ERROR:
+ return "client error";
+ case KVSTORAGE_ERROR_INTERNAL_ERROR:
+ return "library error";
+ }
+
+ /* Not reached */
+ return "unknown error";
+}
#define LIBKVSTORAGECLIENT_H_
#include <glib.h>
+#include <sys/time.h>
/* Errors */
enum rspamd_kvstorage_error {
/* Callbacks for async API */
typedef void (*kvstorage_connect_cb) (enum rspamd_kvstorage_error code,
struct rspamd_kvstorage_connection *conn, gpointer user_data);
-typedef void (*kvstorage_read_cb) (enum rspamd_kvstorage_error code, const gpointer key,
+typedef void (*kvstorage_read_cb) (enum rspamd_kvstorage_error code, const gpointer key, guint keylen,
const gpointer value, gsize datalen, struct rspamd_kvstorage_connection *conn,
gpointer user_data);
-typedef void (*kvstorage_write_cb) (enum rspamd_kvstorage_error code, const gpointer key, struct rspamd_kvstorage_connection *conn,
+typedef void (*kvstorage_write_cb) (enum rspamd_kvstorage_error code, const gpointer key, guint keylen,
+ struct rspamd_kvstorage_connection *conn,
gpointer user_data);
/* Asynced API */
* @param ud user data for callback
*/
enum rspamd_kvstorage_error rspamd_kvstorage_get_async (struct rspamd_kvstorage_connection *conn,
- const gpointer key, kvstorage_read_cb cb, gpointer ud);
+ const gpointer key, guint keylen, kvstorage_read_cb cb, gpointer ud);
/**
* Write key asynced
* @param ud user data for callback
*/
enum rspamd_kvstorage_error rspamd_kvstorage_set_async (struct rspamd_kvstorage_connection *conn,
- const gpointer key, const gpointer value, gsize len, guint expire, kvstorage_write_cb cb, gpointer ud);
+ const gpointer key, guint keylen, const gpointer value, gsize len, guint expire, kvstorage_write_cb cb, gpointer ud);
/**
* Delete key asynced
* @param ud user data for callback
*/
enum rspamd_kvstorage_error rspamd_kvstorage_delete_async (struct rspamd_kvstorage_connection *conn,
- const gpointer key, kvstorage_write_cb cb, gpointer ud);
+ const gpointer key, guint keylen, kvstorage_write_cb cb, gpointer ud);
/**
* Close connection
* @param value value readed
*/
enum rspamd_kvstorage_error rspamd_kvstorage_get_sync (struct rspamd_kvstorage_connection *conn,
- const gpointer key, gpointer **value, guint *len);
+ const gpointer key, guint keylen, gpointer **value, guint *len);
/**
* Write key synced
* @param value data to write
*/
enum rspamd_kvstorage_error rspamd_kvstorage_set_sync (struct rspamd_kvstorage_connection *conn,
- const gpointer key, const gpointer value, gsize len, guint expire);
+ const gpointer key, guint keylen, const gpointer value, gsize len, guint expire);
/**
* Delete key synced
* @param key key to delete
*/
enum rspamd_kvstorage_error rspamd_kvstorage_delete_sync (struct rspamd_kvstorage_connection *conn,
- const gpointer key);
+ const gpointer key, guint keylen);
/**
* Close connection
*/
enum rspamd_kvstorage_error rspamd_kvstorage_close_sync (struct rspamd_kvstorage_connection *conn);
+/**
+ * Convert error code to string
+ * @param err error code to be converted
+ */
+const gchar* rspamd_kvstorage_strerror (enum rspamd_kvstorage_error err);
+
#endif /* LIBKVSTORAGECLIENT_H_ */
*/
extern struct rspamd_main *rspamd_main;
+/* Worker task manipulations */
+
+/**
+ * Construct new task for worker
+ */
+struct worker_task* construct_task (struct rspamd_worker *worker);
+/**
+ * Destroy task object and remove its IO dispatcher if it exists
+ */
+void free_task (struct worker_task *task, gboolean is_soft);
+void free_task_hard (gpointer ud);
+void free_task_soft (gpointer ud);
+
+/**
+ * Set counter for a symbol
+ */
+double set_counter (const gchar *name, guint32 value);
+
#endif
/*
msg_info ("invalidate of statfile pool is planned in %d seconds", (gint)pool->invalidate_tv.tv_sec);
}
}
+
+
+stat_file_t *
+get_statfile_by_symbol (statfile_pool_t *pool, struct classifier_config *ccf,
+ const gchar *symbol, struct statfile **st, gboolean try_create)
+{
+ stat_file_t *res = NULL;
+ GList *cur;
+
+ if (pool == NULL || ccf == NULL || symbol == NULL) {
+ msg_err ("invalid input arguments");
+ return NULL;
+ }
+
+ cur = g_list_first (ccf->statfiles);
+ while (cur) {
+ *st = cur->data;
+ if (strcmp (symbol, (*st)->symbol) == 0) {
+ break;
+ }
+ *st = NULL;
+ cur = g_list_next (cur);
+ }
+ if (*st == NULL) {
+ msg_info ("cannot find statfile with symbol %s", symbol);
+ return NULL;
+ }
+
+ if ((res = statfile_pool_is_open (pool, (*st)->path)) == NULL) {
+ if ((res = statfile_pool_open (pool, (*st)->path, (*st)->size, FALSE)) == NULL) {
+ msg_warn ("cannot open %s", (*st)->path);
+ if (try_create) {
+ if (statfile_pool_create (pool, (*st)->path, (*st)->size) == -1) {
+ msg_err ("cannot create statfile %s", (*st)->path);
+ return NULL;
+ }
+ res = statfile_pool_open (pool, (*st)->path, (*st)->size, FALSE);
+ if (res == NULL) {
+ msg_err ("cannot open statfile %s after creation", (*st)->path);
+ }
+ }
+ }
+ }
+
+ return res;
+}
+
+
struct timeval invalidate_tv;
} statfile_pool_t;
+/* Forwarded declarations */
+struct classifier_config;
+struct statfile;
+
/**
* Create new statfile pool
* @param max_size maximum size
*/
void statfile_pool_plan_invalidate (statfile_pool_t *pool, time_t seconds, time_t jitter);
+/**
+ * Get a statfile by symbol
+ * @param pool pool object
+ * @param ccf ccf classifier config
+ * @param symbol symbol to search
+ * @param st statfile to get
+ * @param try_create whether we need to create statfile if it is absent
+ */
+stat_file_t* get_statfile_by_symbol (statfile_pool_t *pool, struct classifier_config *ccf,
+ const gchar *symbol, struct statfile **st, gboolean try_create);
+
#endif
/* Default connect timeout for sync sockets */
#define CONNECT_TIMEOUT 3
-rspamd_hash_t *counters = NULL;
-
static gchar* rspamd_sprintf_num (gchar *buf, gchar *last, guint64 ui64, gchar zero, guint hexadecimal, guint width);
gint
return (const gchar *)res;
}
-double
-set_counter (const gchar *name, guint32 value)
-{
- struct counter_data *cd;
- double alpha;
- gchar *key;
-
- cd = rspamd_hash_lookup (counters, (gpointer) name);
-
- if (cd == NULL) {
- cd = memory_pool_alloc_shared (counters->pool, sizeof (struct counter_data));
- cd->value = value;
- cd->number = 0;
- key = memory_pool_strdup_shared (counters->pool, name);
- rspamd_hash_insert (counters, (gpointer) key, (gpointer) cd);
- }
- else {
- /* Calculate new value */
- memory_pool_wlock_rwlock (counters->lock);
-
- alpha = 2. / (++cd->number + 1);
- cd->value = cd->value * (1. - alpha) + value * alpha;
-
- memory_pool_wunlock_rwlock (counters->lock);
- }
-
- return cd->value;
-}
-
#ifndef g_tolower
# define g_tolower(x) (((x) >= 'A' && (x) <= 'Z') ? (x) - 'A' + 'a' : (x))
#endif
}
#endif /* HAVE_FLOCK */
-stat_file_t *
-get_statfile_by_symbol (statfile_pool_t *pool, struct classifier_config *ccf,
- const gchar *symbol, struct statfile **st, gboolean try_create)
-{
- stat_file_t *res = NULL;
- GList *cur;
-
- if (pool == NULL || ccf == NULL || symbol == NULL) {
- msg_err ("invalid input arguments");
- return NULL;
- }
-
- cur = g_list_first (ccf->statfiles);
- while (cur) {
- *st = cur->data;
- if (strcmp (symbol, (*st)->symbol) == 0) {
- break;
- }
- *st = NULL;
- cur = g_list_next (cur);
- }
- if (*st == NULL) {
- msg_info ("cannot find statfile with symbol %s", symbol);
- return NULL;
- }
-
- if ((res = statfile_pool_is_open (pool, (*st)->path)) == NULL) {
- if ((res = statfile_pool_open (pool, (*st)->path, (*st)->size, FALSE)) == NULL) {
- msg_warn ("cannot open %s", (*st)->path);
- if (try_create) {
- if (statfile_pool_create (pool, (*st)->path, (*st)->size) == -1) {
- msg_err ("cannot create statfile %s", (*st)->path);
- return NULL;
- }
- res = statfile_pool_open (pool, (*st)->path, (*st)->size, FALSE);
- if (res == NULL) {
- msg_err ("cannot open statfile %s after creation", (*st)->path);
- }
- }
- }
- }
-
- return res;
-}
-
-
#if ((GLIB_MAJOR_VERSION == 2) && (GLIB_MINOR_VERSION < 22))
void
return TYPE_UNKNOWN;
}
-/*
- * Destructor for recipients list
- */
-static void
-rcpt_destruct (void *pointer)
-{
- struct worker_task *task = (struct worker_task *) pointer;
-
- if (task->rcpt) {
- g_list_free (task->rcpt);
- }
-}
-
-
/* Compare two emails for building emails tree */
-static gint
+gint
compare_email_func (gconstpointer a, gconstpointer b)
{
const struct uri *u1 = a, *u2 = b;
return 0;
}
-static gint
+gint
compare_url_func (gconstpointer a, gconstpointer b)
{
const struct uri *u1 = a, *u2 = b;
return r;
}
-/*
- * Create new task
- */
-struct worker_task *
-construct_task (struct rspamd_worker *worker)
-{
- struct worker_task *new_task;
-
- new_task = g_slice_alloc0 (sizeof (struct worker_task));
-
- new_task->worker = worker;
- new_task->state = READ_COMMAND;
- new_task->cfg = worker->srv->cfg;
- new_task->from_addr.s_addr = INADDR_NONE;
- new_task->view_checked = FALSE;
-#ifdef HAVE_CLOCK_GETTIME
-# ifdef HAVE_CLOCK_PROCESS_CPUTIME_ID
- clock_gettime (CLOCK_PROCESS_CPUTIME_ID, &new_task->ts);
-# elif defined(HAVE_CLOCK_VIRTUAL)
- clock_gettime (CLOCK_VIRTUAL, &new_task->ts);
-# else
- clock_gettime (CLOCK_REALTIME, &new_task->ts);
-# endif
-#endif
- if (gettimeofday (&new_task->tv, NULL) == -1) {
- msg_warn ("gettimeofday failed: %s", strerror (errno));
- }
-
- new_task->task_pool = memory_pool_new (memory_pool_get_size ());
-
- /* Add destructor for recipients list (it would be better to use anonymous function here */
- memory_pool_add_destructor (new_task->task_pool,
- (pool_destruct_func) rcpt_destruct, new_task);
- new_task->results = g_hash_table_new (g_str_hash, g_str_equal);
- memory_pool_add_destructor (new_task->task_pool,
- (pool_destruct_func) g_hash_table_destroy,
- new_task->results);
- new_task->re_cache = g_hash_table_new (g_str_hash, g_str_equal);
- memory_pool_add_destructor (new_task->task_pool,
- (pool_destruct_func) g_hash_table_destroy,
- new_task->re_cache);
- new_task->raw_headers = g_hash_table_new (rspamd_strcase_hash, rspamd_strcase_equal);
- memory_pool_add_destructor (new_task->task_pool,
- (pool_destruct_func) g_hash_table_destroy,
- new_task->raw_headers);
- new_task->emails = g_tree_new (compare_email_func);
- memory_pool_add_destructor (new_task->task_pool,
- (pool_destruct_func) g_tree_destroy,
- new_task->emails);
- new_task->urls = g_tree_new (compare_url_func);
- memory_pool_add_destructor (new_task->task_pool,
- (pool_destruct_func) g_tree_destroy,
- new_task->urls);
- new_task->s =
- new_async_session (new_task->task_pool, free_task_hard, new_task);
- new_task->sock = -1;
- new_task->is_mime = TRUE;
-
- return new_task;
-}
-
-
-/*
- * Free all structures of worker_task
- */
-void
-free_task (struct worker_task *task, gboolean is_soft)
-{
- GList *part;
- struct mime_part *p;
-
- if (task) {
- debug_task ("free pointer %p", task);
- while ((part = g_list_first (task->parts))) {
- task->parts = g_list_remove_link (task->parts, part);
- p = (struct mime_part *) part->data;
- g_byte_array_free (p->content, TRUE);
- g_list_free_1 (part);
- }
- if (task->text_parts) {
- g_list_free (task->text_parts);
- }
- if (task->images) {
- g_list_free (task->images);
- }
- if (task->messages) {
- g_list_free (task->messages);
- }
- if (task->received) {
- g_list_free (task->received);
- }
- memory_pool_delete (task->task_pool);
- if (task->dispatcher) {
- if (is_soft) {
- /* Plan dispatcher shutdown */
- task->dispatcher->wanna_die = 1;
- }
- else {
- rspamd_remove_dispatcher (task->dispatcher);
- }
- }
- if (task->sock != -1) {
- close (task->sock);
- }
- g_slice_free1 (sizeof (struct worker_task), task);
- }
-}
-
-void
-free_task_hard (gpointer ud)
-{
- struct worker_task *task = ud;
-
- free_task (task, FALSE);
-}
-
-void
-free_task_soft (gpointer ud)
-{
- struct worker_task *task = ud;
-
- free_task (task, FALSE);
-}
-
gchar *
escape_braces_addr_fstr (memory_pool_t *pool, f_str_t *in)
{
const gchar* calculate_check_time (struct timeval *begin, gint resolution);
#endif
-/*
- * Set counter for a symbol
- */
-double set_counter (const gchar *name, guint32 value);
-
/*
* File locking functions
*/
*/
void gperf_profiler_init (struct config_file *cfg, const gchar *descr);
-/*
- * Get a statfile by symbol
- */
-stat_file_t* get_statfile_by_symbol (statfile_pool_t *pool, struct classifier_config *ccf,
- const gchar *symbol, struct statfile **st, gboolean try_create);
-
/*
* Workaround for older versions of glib
*/
#define msec_to_tv(msec, tv) do { (tv)->tv_sec = (msec) / 1000; (tv)->tv_usec = ((msec) - (tv)->tv_sec * 1000) * 1000; } while(0)
#define tv_to_msec(tv) (tv)->tv_sec * 1000 + (tv)->tv_usec / 1000
-struct worker_task;
-struct rspamd_worker;
+/* Compare two emails for building emails tree */
+gint compare_email_func (gconstpointer a, gconstpointer b);
-/**
- * Construct new task for worker
- */
-struct worker_task* construct_task (struct rspamd_worker *worker);
-/**
- * Destroy task object and remove its IO dispatcher if it exists
- */
-void free_task (struct worker_task *task, gboolean is_soft);
-void free_task_hard (gpointer ud);
-void free_task_soft (gpointer ud);
+/* Compare two urls for building emails tree */
+gint compare_url_func (gconstpointer a, gconstpointer b);
/*
* Find string find in string s ignoring case
static sig_atomic_t wanna_die = 0;
+rspamd_hash_t *counters = NULL;
+
#ifndef HAVE_SA_SIGINFO
static void
sig_handler (gint signo)
return;
}
+/*
+ * Destructor for recipients list in a task
+ */
+static void
+rcpt_destruct (void *pointer)
+{
+ struct worker_task *task = (struct worker_task *) pointer;
+
+ if (task->rcpt) {
+ g_list_free (task->rcpt);
+ }
+}
+
+/*
+ * Create new task
+ */
+struct worker_task *
+construct_task (struct rspamd_worker *worker)
+{
+ struct worker_task *new_task;
+
+ new_task = g_slice_alloc0 (sizeof (struct worker_task));
+
+ new_task->worker = worker;
+ new_task->state = READ_COMMAND;
+ new_task->cfg = worker->srv->cfg;
+ new_task->from_addr.s_addr = INADDR_NONE;
+ new_task->view_checked = FALSE;
+#ifdef HAVE_CLOCK_GETTIME
+# ifdef HAVE_CLOCK_PROCESS_CPUTIME_ID
+ clock_gettime (CLOCK_PROCESS_CPUTIME_ID, &new_task->ts);
+# elif defined(HAVE_CLOCK_VIRTUAL)
+ clock_gettime (CLOCK_VIRTUAL, &new_task->ts);
+# else
+ clock_gettime (CLOCK_REALTIME, &new_task->ts);
+# endif
+#endif
+ if (gettimeofday (&new_task->tv, NULL) == -1) {
+ msg_warn ("gettimeofday failed: %s", strerror (errno));
+ }
+
+ new_task->task_pool = memory_pool_new (memory_pool_get_size ());
+
+ /* Add destructor for recipients list (it would be better to use anonymous function here */
+ memory_pool_add_destructor (new_task->task_pool,
+ (pool_destruct_func) rcpt_destruct, new_task);
+ new_task->results = g_hash_table_new (g_str_hash, g_str_equal);
+ memory_pool_add_destructor (new_task->task_pool,
+ (pool_destruct_func) g_hash_table_destroy,
+ new_task->results);
+ new_task->re_cache = g_hash_table_new (g_str_hash, g_str_equal);
+ memory_pool_add_destructor (new_task->task_pool,
+ (pool_destruct_func) g_hash_table_destroy,
+ new_task->re_cache);
+ new_task->raw_headers = g_hash_table_new (rspamd_strcase_hash, rspamd_strcase_equal);
+ memory_pool_add_destructor (new_task->task_pool,
+ (pool_destruct_func) g_hash_table_destroy,
+ new_task->raw_headers);
+ new_task->emails = g_tree_new (compare_email_func);
+ memory_pool_add_destructor (new_task->task_pool,
+ (pool_destruct_func) g_tree_destroy,
+ new_task->emails);
+ new_task->urls = g_tree_new (compare_url_func);
+ memory_pool_add_destructor (new_task->task_pool,
+ (pool_destruct_func) g_tree_destroy,
+ new_task->urls);
+ new_task->s =
+ new_async_session (new_task->task_pool, free_task_hard, new_task);
+ new_task->sock = -1;
+ new_task->is_mime = TRUE;
+
+ return new_task;
+}
+
+
+/*
+ * Free all structures of worker_task
+ */
+void
+free_task (struct worker_task *task, gboolean is_soft)
+{
+ GList *part;
+ struct mime_part *p;
+
+ if (task) {
+ debug_task ("free pointer %p", task);
+ while ((part = g_list_first (task->parts))) {
+ task->parts = g_list_remove_link (task->parts, part);
+ p = (struct mime_part *) part->data;
+ g_byte_array_free (p->content, TRUE);
+ g_list_free_1 (part);
+ }
+ if (task->text_parts) {
+ g_list_free (task->text_parts);
+ }
+ if (task->images) {
+ g_list_free (task->images);
+ }
+ if (task->messages) {
+ g_list_free (task->messages);
+ }
+ if (task->received) {
+ g_list_free (task->received);
+ }
+ memory_pool_delete (task->task_pool);
+ if (task->dispatcher) {
+ if (is_soft) {
+ /* Plan dispatcher shutdown */
+ task->dispatcher->wanna_die = 1;
+ }
+ else {
+ rspamd_remove_dispatcher (task->dispatcher);
+ }
+ }
+ if (task->sock != -1) {
+ close (task->sock);
+ }
+ g_slice_free1 (sizeof (struct worker_task), task);
+ }
+}
+
+void
+free_task_hard (gpointer ud)
+{
+ struct worker_task *task = ud;
+
+ free_task (task, FALSE);
+}
+
+void
+free_task_soft (gpointer ud)
+{
+ struct worker_task *task = ud;
+
+ free_task (task, FALSE);
+}
+
+double
+set_counter (const gchar *name, guint32 value)
+{
+ struct counter_data *cd;
+ double alpha;
+ gchar *key;
+
+ cd = rspamd_hash_lookup (counters, (gpointer) name);
+
+ if (cd == NULL) {
+ cd = memory_pool_alloc_shared (counters->pool, sizeof (struct counter_data));
+ cd->value = value;
+ cd->number = 0;
+ key = memory_pool_strdup_shared (counters->pool, name);
+ rspamd_hash_insert (counters, (gpointer) key, (gpointer) cd);
+ }
+ else {
+ /* Calculate new value */
+ memory_pool_wlock_rwlock (counters->lock);
+
+ alpha = 2. / (++cd->number + 1);
+ cd->value = cd->value * (1. - alpha) + value * alpha;
+
+ memory_pool_wunlock_rwlock (counters->lock);
+ }
+
+ return cd->value;
+}
+
#ifndef BUILD_STATIC
static void
fin_custom_filters (struct worker_task *task)
}
#endif
+
/*
* Callback that is called when there is data to read in buffer
*/
projects / rspamd.git / commitdiff