#include "config.h"
#include "binlog.h"
+#include "cfg_file.h"
#include "tokenizers/tokenizers.h"
#define BINLOG_SUFFIX ".binlog"
#define VALID_MAGIC { 'r', 's', 'l' }
#define VALID_VERSION { '1', '0' }
+static GHashTable *binlog_opened = NULL;
+static memory_pool_t *binlog_pool = NULL;
+
static gboolean
binlog_write_header (struct rspamd_binlog *log)
{
unlock_file (log->fd, FALSE);
return FALSE;
}
- g_free (log->cur_idx);
- /* Now mmap it to memory */
- if ((log->cur_idx = mmap (NULL, sizeof (struct rspamd_index_block),
- PROT_READ | PROT_WRITE, MAP_SHARED, log->fd, log->metaindex->indexes[0])) == MAP_FAILED) {
- msg_warn ("binlog_write_header: cannot mmap file %s, error %d, %s", log->filename, errno, strerror (errno));
- unlock_file (log->fd, FALSE);
- return FALSE;
- }
unlock_file (log->fd, FALSE);
return FALSE;
}
/* Now mmap metaindex and current index */
- if ((log->metaindex = mmap (NULL, sizeof (struct rspamd_binlog_metaindex),
- PROT_READ | PROT_WRITE, MAP_SHARED, log->fd, sizeof (struct rspamd_binlog_header))) == MAP_FAILED) {
- msg_warn ("binlog_check file: cannot mmap file %s, error %d, %s", log->filename, errno, strerror (errno));
+ if (log->metaindex == NULL) {
+ log->metaindex = g_malloc (sizeof (struct rspamd_binlog_metaindex));
+ }
+ if ((read (log->fd, log->metaindex, sizeof (struct rspamd_binlog_metaindex))) != sizeof (struct rspamd_binlog_metaindex)) {
+ msg_warn ("binlog_check file: cannot read metaindex of file %s, error %d, %s", log->filename, errno, strerror (errno));
return FALSE;
}
/* Current index */
- if ((log->cur_idx = mmap (NULL, sizeof (struct rspamd_index_block),
- PROT_READ | PROT_WRITE, MAP_SHARED, log->fd, log->metaindex->indexes[log->metaindex->last_index])) == MAP_FAILED) {
- msg_warn ("binlog_check_file: cannot mmap file %s, error %d, %s", log->filename, errno, strerror (errno));
+ if (log->cur_idx == NULL) {
+ log->cur_idx = g_malloc (sizeof (struct rspamd_index_block));
+ }
+ if (lseek (log->fd, log->metaindex->indexes[log->metaindex->last_index], SEEK_SET) == -1) {
+ msg_info ("binlog_check_file: cannot seek in file: %s, error: %s", log->filename, strerror (errno));
+ return FALSE;
+ }
+ if ((read (log->fd, log->cur_idx, sizeof (struct rspamd_index_block))) != sizeof (struct rspamd_index_block)) {
+ msg_warn ("binlog_check_file: cannot read index in file %s, error %d, %s", log->filename, errno, strerror (errno));
return FALSE;
}
new = memory_pool_alloc0 (pool, sizeof (struct rspamd_binlog));
new->pool = pool;
new->rotate_time = rotate_time;
- new->rotate_jitter = g_random_int_range (0, rotate_jitter);
new->fd = -1;
+
+ if (rotate_time) {
+ new->rotate_jitter = g_random_int_range (0, rotate_jitter);
+ }
new->filename = memory_pool_alloc (pool, len + sizeof (BINLOG_SUFFIX));
g_strlcpy (new->filename, path, len + 1);
{
if (log) {
if (log->metaindex) {
- munmap (log->metaindex, sizeof (struct rspamd_binlog_metaindex));
+ g_free (log->metaindex);
}
if (log->cur_idx) {
- munmap (log->cur_idx, sizeof (struct rspamd_index_block));
+ g_free (log->cur_idx);
}
close (log->fd);
}
{
off_t seek;
struct rspamd_binlog_index *idx;
-
- /* Seek to end of file */
- if ((seek = lseek (log->fd, 0, SEEK_END)) == -1) {
+
+ lock_file (log->fd, FALSE);
+ /* Write index */
+ log->cur_idx->last_index ++;
+ if (lseek (log->fd, log->metaindex->indexes[log->metaindex->last_index] + G_STRUCT_OFFSET (struct rspamd_index_block, last_index), SEEK_SET) == -1) {
+ unlock_file (log->fd, FALSE);
msg_info ("binlog_insert: cannot seek in file: %s, error: %s", log->filename, strerror (errno));
return FALSE;
}
+ if (write (log->fd, &log->cur_idx->last_index, sizeof (log->cur_idx->last_index)) == -1) {
+ unlock_file (log->fd, FALSE);
+ msg_info ("binlog_insert: cannot write index to file: %s, error: %s", log->filename, strerror (errno));
+ return FALSE;
+ }
- lock_file (log->fd, FALSE);
- /* Write index */
- log->cur_idx->last_index ++;
log->cur_seq ++;
idx = &log->cur_idx->indexes[log->cur_idx->last_index];
idx->seek = seek;
idx->time = (uint64_t)time (NULL);
idx->len = g_tree_nnodes (nodes) * sizeof (struct rspamd_binlog_element);
+ if (lseek (log->fd, log->metaindex->indexes[log->metaindex->last_index] +
+ log->cur_idx->last_index * sizeof (struct rspamd_binlog_index), SEEK_SET) == -1) {
+ unlock_file (log->fd, FALSE);
+ msg_info ("binlog_insert: cannot seek in file: %s, error: %s", log->filename, strerror (errno));
+ return FALSE;
+ }
+ if (write (log->fd, idx, sizeof (struct rspamd_binlog_index)) == -1) {
+ unlock_file (log->fd, FALSE);
+ msg_info ("binlog_insert: cannot write index to file: %s, error: %s", log->filename, strerror (errno));
+ return FALSE;
+ }
+
+ /* Seek to end of file */
+ if ((seek = lseek (log->fd, 0, SEEK_END)) == -1) {
+ unlock_file (log->fd, FALSE);
+ msg_info ("binlog_insert: cannot seek in file: %s, error: %s", log->filename, strerror (errno));
+ return FALSE;
+ }
/* Now write all nodes to file */
g_tree_foreach (nodes, binlog_tree_callback, (gpointer)log);
{
off_t seek;
- /* Seek to end of file */
- if ((seek = lseek (log->fd, 0, SEEK_END)) == -1) {
- msg_info ("create_new_metaindex_block: cannot seek in file: %s, error: %s", log->filename, strerror (errno));
- return FALSE;
- }
lock_file (log->fd, FALSE);
log->metaindex->last_index ++;
/* Offset to metaindex */
log->metaindex->indexes[log->metaindex->last_index] = seek;
- munmap (log->cur_idx, sizeof (struct rspamd_index_block));
- /* Alloc, write, mmap */
- log->cur_idx = g_malloc (sizeof (struct rspamd_index_block));
+ /* Overwrite all metaindexes */
+ if (lseek (log->fd, sizeof (struct rspamd_binlog_header), SEEK_SET) == -1) {
+ unlock_file (log->fd, FALSE);
+ msg_info ("create_new_metaindex_block: cannot seek in file: %s, error: %s", log->filename, strerror (errno));
+ return FALSE;
+ }
+ if (write (log->fd, log->metaindex, sizeof (struct rspamd_binlog_metaindex)) == -1) {
+ unlock_file (log->fd, FALSE);
+ msg_info ("create_new_metaindex_block: cannot write metaindex in file: %s, error: %s", log->filename, strerror (errno));
+ return FALSE;
+ }
bzero (log->cur_idx, sizeof (struct rspamd_index_block));
- if (write (log->fd, log->cur_idx, sizeof (struct rspamd_index_block)) == -1) {
+ /* Seek to end of file */
+ if ((seek = lseek (log->fd, 0, SEEK_END)) == -1) {
unlock_file (log->fd, FALSE);
- g_free (log->cur_idx);
- msg_warn ("create_new_metaindex_block: cannot write file %s, error %d, %s", log->filename, errno, strerror (errno));
+ msg_info ("create_new_metaindex_block: cannot seek in file: %s, error: %s", log->filename, strerror (errno));
return FALSE;
}
- g_free (log->cur_idx);
- /* Now mmap it to memory */
- if ((log->cur_idx = mmap (NULL, sizeof (struct rspamd_index_block),
- PROT_READ | PROT_WRITE, MAP_SHARED, log->fd, seek)) == MAP_FAILED) {
- log->cur_idx = NULL;
+ if (write (log->fd, log->cur_idx, sizeof (struct rspamd_index_block)) == -1) {
unlock_file (log->fd, FALSE);
- msg_warn ("create_new_metaindex_block: cannot mmap file %s, error %d, %s", log->filename, errno, strerror (errno));
+ g_free (log->cur_idx);
+ msg_warn ("create_new_metaindex_block: cannot write file %s, error %d, %s", log->filename, errno, strerror (errno));
return FALSE;
}
unlock_file (log->fd, FALSE);
{
uint64_t now = time (NULL);
- if ((now - log->header.create_time) > log->rotate_time + log->rotate_jitter) {
+ if (log->rotate_time && ((now - log->header.create_time) > log->rotate_time + log->rotate_jitter)) {
return TRUE;
}
return FALSE;
/* Unmap mapped fragments */
if (log->metaindex) {
- munmap (log->metaindex, sizeof (struct rspamd_binlog_metaindex));
+ g_free (log->metaindex);
log->metaindex = NULL;
}
if (log->cur_idx) {
- munmap (log->cur_idx, sizeof (struct rspamd_index_block));
+ g_free (log->cur_idx);
log->cur_idx = NULL;
}
/* Format backup name */
}
else {
/* Unmap old fragment */
- munmap ((*rep)->data, (*rep)->len);
+ g_free ((*rep)->data);
}
if (from_rev == log->cur_seq) {
else if (metaindex_num != log->metaindex->last_index) {
/* Need to remap index block */
lock_file (log->fd, FALSE);
- if ((idxb = mmap (NULL, sizeof (struct rspamd_index_block),
- PROT_READ | PROT_WRITE, MAP_SHARED,
- log->fd, log->metaindex->indexes[metaindex_num])) == MAP_FAILED) {
+ idxb = g_malloc (sizeof (struct rspamd_index_block));
+ idx_mapped = TRUE;
+ if (lseek (log->fd, log->metaindex->indexes[metaindex_num], SEEK_SET) == -1) {
unlock_file (log->fd, FALSE);
- msg_warn ("binlog_sync: cannot mmap file %s, error %d, %s", log->filename, errno, strerror (errno));
- return FALSE;
+ msg_warn ("binlog_sync: cannot seek file %s, error %d, %s", log->filename, errno, strerror (errno));
+ res = FALSE;
+ goto end;
}
- idx_mapped = TRUE;
+ if ((read (log->fd, &idxb, sizeof (struct rspamd_index_block))) != sizeof (struct rspamd_index_block)) {
+ unlock_file (log->fd, FALSE);
+ msg_warn ("binlog_sync: cannot read index from file %s, error %d, %s", log->filename, errno, strerror (errno));
+ res = FALSE;
+ goto end;
+ }
+ unlock_file (log->fd, FALSE);
}
else {
idxb = log->cur_idx;
/* Now fill reply structure */
(*rep)->len = idx->len;
- /* MMap result */
- if (((*rep)->data = mmap (NULL, idx->len, PROT_READ, MAP_SHARED, log->fd, idx->seek)) == MAP_FAILED) {
- msg_warn ("binlog_sync: cannot mmap file %s, error %d, %s", log->filename, errno, strerror (errno));
+ /* Read result */
+ if (lseek (log->fd, idx->seek, SEEK_SET) == -1) {
+ msg_warn ("binlog_sync: cannot seek file %s, error %d, %s", log->filename, errno, strerror (errno));
+ res = FALSE;
+ goto end;
+ }
+
+ (*rep)->data = g_malloc (idx->len);
+ if ((read (log->fd, (*rep)->data, idx->len)) != idx->len) {
+ msg_warn ("binlog_sync: cannot read file %s, error %d, %s", log->filename, errno, strerror (errno));
res = FALSE;
goto end;
}
end:
if (idx_mapped) {
- munmap (idxb, sizeof (struct rspamd_index_block));
+ g_free (idxb);
}
return res;
}
+static gboolean
+maybe_init_static ()
+{
+ if (!binlog_opened) {
+ binlog_opened = g_hash_table_new (g_direct_hash, g_direct_equal);
+ if (!binlog_opened) {
+ return FALSE;
+ }
+ }
+
+ if (!binlog_pool) {
+ binlog_pool = memory_pool_new (memory_pool_get_size ());
+ if (!binlog_pool) {
+ return FALSE;
+ }
+ }
+
+ return TRUE;
+}
+
+void
+maybe_write_binlog (struct classifier_config *ccf, const char *symbol, GTree *nodes)
+{
+ struct rspamd_binlog *log;
+ struct statfile *st = NULL;
+ GList *cur;
+
+ if (ccf == NULL) {
+ return;
+ }
+
+ 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 || nodes == NULL || st->binlog == NULL || st->binlog->affinity != AFFINITY_MASTER) {
+ return;
+ }
+
+ if (!maybe_init_static ()) {
+ return;
+ }
+
+ if ((log = g_hash_table_lookup (binlog_opened, st)) == NULL) {
+ if ((log = binlog_open (binlog_pool, st->path, st->binlog->rotate_time, st->binlog->rotate_time / 2)) != NULL) {
+ g_hash_table_insert (binlog_opened, st, log);
+ }
+ else {
+ return;
+ }
+ }
+
+ (void)binlog_insert (log, nodes);
+}
+
+
int
add_memcached_server (struct config_file *cf, char *str)
{
- char *cur_tok, *err_str;
struct memcached_server *mc;
- struct hostent *hent;
uint16_t port;
if (str == NULL)
return 0;
- cur_tok = strsep (&str, ":");
-
- if (cur_tok == NULL || *cur_tok == '\0')
- return 0;
-
if (cf->memcached_servers_num == MAX_MEMCACHED_SERVERS) {
yywarn ("yyparse: maximum number of memcached servers is reached %d", MAX_MEMCACHED_SERVERS);
+ return 0;
}
mc = &cf->memcached_servers[cf->memcached_servers_num];
- if (mc == NULL)
- return 0;
/* cur_tok - server name, str - server port */
- if (str == NULL) {
- port = DEFAULT_MEMCACHED_PORT;
- }
- else {
- port = (uint16_t) strtoul (str, &err_str, 10);
- if (*err_str != '\0') {
- return 0;
- }
+ port = DEFAULT_MEMCACHED_PORT;
+
+ if (!parse_host_port (str, &mc->addr, &port)) {
+ return 0;
}
- if (!inet_aton (cur_tok, &mc->addr)) {
+ mc->port = port;
+ cf->memcached_servers_num++;
+ return 1;
+}
+
+gboolean
+parse_host_port (const char *str, struct in_addr *ina, uint16_t *port)
+{
+ char **tokens, *err_str;
+ struct hostent *hent;
+ unsigned int port_parsed, saved_errno = errno;
+
+ tokens = g_strsplit (str, ":", 0);
+ if (!tokens || !tokens[0]) {
+ return FALSE;
+ }
+
+ /* Now try to parse host and write address to ina */
+ if (!inet_aton (tokens[0], ina)) {
/* Try to call gethostbyname */
- hent = gethostbyname (cur_tok);
+ hent = gethostbyname (tokens[0]);
if (hent == NULL) {
- return 0;
+ msg_warn ("parse_host_port: cannot resolve %s", tokens[0]);
+ goto err;
}
else {
- memcpy ((char *)&mc->addr, hent->h_addr, sizeof (struct in_addr));
+ memcpy (ina, hent->h_addr, sizeof (struct in_addr));
}
}
- mc->port = port;
- cf->memcached_servers_num++;
- return 1;
+ if (tokens[1] != NULL) {
+ /* Port part */
+ errno = 0;
+ port_parsed = strtoul (tokens[1], &err_str, 10);
+ if (*err_str != '\0' || errno != 0) {
+ msg_warn ("parse_host_port: cannot parse port: %s, at symbol %c, error: %s", tokens[1], *err_str, strerror (errno));
+ goto err;
+ }
+ if (port_parsed > G_MAXUINT16) {
+ errno = ERANGE;
+ msg_warn ("parse_host_port: cannot parse port: %s, error: %s", tokens[1], *err_str, strerror (errno));
+ goto err;
+ }
+ *port = port_parsed;
+ }
+
+ /* Restore errno */
+ errno = saved_errno;
+ g_strfreev (tokens);
+ return TRUE;
+
+err:
+ errno = saved_errno;
+ g_strfreev (tokens);
+ return FALSE;
}
int
parse_bind_line (struct config_file *cfg, struct worker_conf *cf, char *str)
{
- char *cur_tok, *err_str;
- struct hostent *hent;
- size_t s;
char **host;
int16_t *family, *port;
struct in_addr *addr;
if (str == NULL)
return 0;
- cur_tok = strsep (&str, ":");
host = &cf->bind_host;
port = &cf->bind_port;
family = &cf->bind_family;
addr = &cf->bind_addr;
- if (cur_tok[0] == '/' || cur_tok[0] == '.') {
+ if (str[0] == '/' || str[0] == '.') {
#ifdef HAVE_DIRNAME
/* Try to check path of bind credit */
struct stat st;
int fd;
- char *copy = memory_pool_strdup (cfg->cfg_pool, cur_tok);
+ char *copy = memory_pool_strdup (cfg->cfg_pool, str);
if (stat (copy, &st) == -1) {
if (errno == ENOENT) {
- if ((fd = open (cur_tok, O_RDWR | O_TRUNC | O_CREAT, S_IWUSR | S_IRUSR)) == -1) {
- yyerror ("parse_bind_line: cannot open path %s for making socket, %s", cur_tok, strerror (errno));
+ if ((fd = open (str, O_RDWR | O_TRUNC | O_CREAT, S_IWUSR | S_IRUSR)) == -1) {
+ yyerror ("parse_bind_line: cannot open path %s for making socket, %s", str, strerror (errno));
return 0;
}
else {
close (fd);
- unlink (cur_tok);
+ unlink (str);
}
}
else {
- yyerror ("parse_bind_line: cannot stat path %s for making socket, %s", cur_tok, strerror (errno));
+ yyerror ("parse_bind_line: cannot stat path %s for making socket, %s", str, strerror (errno));
return 0;
}
}
else {
- if (unlink (cur_tok) == -1) {
- yyerror ("parse_bind_line: cannot remove path %s for making socket, %s", cur_tok, strerror (errno));
+ if (unlink (str) == -1) {
+ yyerror ("parse_bind_line: cannot remove path %s for making socket, %s", str, strerror (errno));
return 0;
}
}
#endif
- *host = memory_pool_strdup (cfg->cfg_pool, cur_tok);
+ *host = memory_pool_strdup (cfg->cfg_pool, str);
*family = AF_UNIX;
return 1;
-
}
else {
- if (*str != '\0') {
- *port = (uint16_t) strtoul (str, &err_str, 10);
- if (*err_str != '\0') {
- yyerror ("parse_bind_line: cannot read numeric value: %s", err_str);
- return 0;
- }
- }
- if (strcmp (cur_tok, "*") == 0) {
- *host = memory_pool_strdup (cfg->cfg_pool, cur_tok);
- addr->s_addr = htonl (INADDR_ANY);
- }
- else if (!inet_aton (cur_tok, addr)) {
- /* Try to call gethostbyname */
- hent = gethostbyname (cur_tok);
- if (hent == NULL) {
- return 0;
- }
- else {
- *host = memory_pool_strdup (cfg->cfg_pool, cur_tok);
- memcpy ((char *)addr, hent->h_addr, sizeof (struct in_addr));
- s = strlen (cur_tok) + 1;
- }
- }
- else {
- *host = memory_pool_strdup (cfg->cfg_pool, cur_tok);
- }
- *family = AF_INET;
+ if (parse_host_port (str, addr, port)) {
+ *host = memory_pool_strdup (cfg->cfg_pool, str);
+ *family = AF_INET;
- return 1;
+ return 1;
+ }
}
return 0;
projects / rspamd.git / commitdiff