RuhNetConsulting
/
rtpengine
mirror of https://github.com/sipwise/rtpengine
1
1
Fork 0
Code Issues 0 Projects 0 Releases 822 Wiki Activity
Browse Source

remove redis-role redundancy 

Change-Id: Id5a39d008169d4b01cd33a7d82bb0896983e47bc
changes/96/4596/1
Richard Fuchs 10 years ago
parent
27d18e2be3
commit
725638bccb
6 changed files with 49 additions and 45 deletions
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +9
    -9
      daemon/call.c
  2. +6
    -6
      daemon/call_interfaces.c
  3. +2
    -2
      daemon/main.c
  4. +2
    -2
      daemon/media_socket.c
  5. +19
    -18
      daemon/redis.c
  6. +11
    -8
      daemon/redis.h

+ 9
- 9
daemon/call.c View File

@ -568,9 +568,9 @@ static void callmaster_timer(void *ptr) {
if (update) { if (update) {
if (m->conf.redis_write) { if (m->conf.redis_write) {
redis_update(ps->call, m->conf.redis_write, ANY_REDIS_ROLE);
redis_update(ps->call, m->conf.redis_write);
} else if (m->conf.redis) { } else if (m->conf.redis) {
redis_update(ps->call, m->conf.redis, MASTER_REDIS_ROLE);
redis_update(ps->call, m->conf.redis);
} }
} }
@ -1847,9 +1847,9 @@ void call_destroy(struct call *c) {
obj_put(c); obj_put(c);
if (m->conf.redis_write) { if (m->conf.redis_write) {
redis_delete(c, m->conf.redis_write, ANY_REDIS_ROLE);
redis_delete(c, m->conf.redis_write);
} else if (m->conf.redis) { } else if (m->conf.redis) {
redis_delete(c, m->conf.redis, MASTER_REDIS_ROLE);
redis_delete(c, m->conf.redis);
} }
rwlock_lock_w(&c->master_lock); rwlock_lock_w(&c->master_lock);
@ -2713,9 +2713,9 @@ static void calls_dump_iterator(void *key, void *val, void *ptr) {
struct callmaster *m = c->callmaster; struct callmaster *m = c->callmaster;
if (m->conf.redis_write) { if (m->conf.redis_write) {
redis_update(c, m->conf.redis_write, ANY_REDIS_ROLE);
redis_update(c, m->conf.redis_write);
} else if (m->conf.redis) { } else if (m->conf.redis) {
redis_update(c, m->conf.redis, MASTER_REDIS_ROLE);
redis_update(c, m->conf.redis);
} }
} }
@ -2724,7 +2724,7 @@ void calls_dump_redis(struct callmaster *m) {
return; return;
ilog(LOG_DEBUG, "Start dumping all call data to Redis...\n"); ilog(LOG_DEBUG, "Start dumping all call data to Redis...\n");
redis_wipe(m->conf.redis, MASTER_REDIS_ROLE);
redis_wipe(m->conf.redis);
g_hash_table_foreach(m->callhash, calls_dump_iterator, NULL); g_hash_table_foreach(m->callhash, calls_dump_iterator, NULL);
ilog(LOG_DEBUG, "Finished dumping all call data to Redis\n"); ilog(LOG_DEBUG, "Finished dumping all call data to Redis\n");
} }
@ -2734,7 +2734,7 @@ void calls_dump_redis_read(struct callmaster *m) {
return; return;
ilog(LOG_DEBUG, "Start dumping all call data to read Redis...\n"); ilog(LOG_DEBUG, "Start dumping all call data to read Redis...\n");
redis_wipe(m->conf.redis_read, ANY_REDIS_ROLE);
redis_wipe(m->conf.redis_read);
g_hash_table_foreach(m->callhash, calls_dump_iterator, NULL); g_hash_table_foreach(m->callhash, calls_dump_iterator, NULL);
ilog(LOG_DEBUG, "Finished dumping all call data to read Redis\n"); ilog(LOG_DEBUG, "Finished dumping all call data to read Redis\n");
} }
@ -2744,7 +2744,7 @@ void calls_dump_redis_write(struct callmaster *m) {
return; return;
ilog(LOG_DEBUG, "Start dumping all call data to write Redis...\n"); ilog(LOG_DEBUG, "Start dumping all call data to write Redis...\n");
redis_wipe(m->conf.redis_write, ANY_REDIS_ROLE);
redis_wipe(m->conf.redis_write);
g_hash_table_foreach(m->callhash, calls_dump_iterator, NULL); g_hash_table_foreach(m->callhash, calls_dump_iterator, NULL);
ilog(LOG_DEBUG, "Finished dumping all call data to write Redis\n"); ilog(LOG_DEBUG, "Finished dumping all call data to write Redis\n");
} }


+ 6
- 6
daemon/call_interfaces.c View File

@ -187,9 +187,9 @@ static str *call_update_lookup_udp(char **out, struct callmaster *m, enum call_o
rwlock_unlock_w(&c->master_lock); rwlock_unlock_w(&c->master_lock);
if (m->conf.redis_write) { if (m->conf.redis_write) {
redis_update(c, m->conf.redis_write, ANY_REDIS_ROLE);
redis_update(c, m->conf.redis_write);
} else if (m->conf.redis) { } else if (m->conf.redis) {
redis_update(c, m->conf.redis, MASTER_REDIS_ROLE);
redis_update(c, m->conf.redis);
} }
gettimeofday(&(monologue->started), NULL); gettimeofday(&(monologue->started), NULL);
@ -339,9 +339,9 @@ out2:
streams_free(&s); streams_free(&s);
if (m->conf.redis_write) { if (m->conf.redis_write) {
redis_update(c, m->conf.redis_write, ANY_REDIS_ROLE);
redis_update(c, m->conf.redis_write);
} else if (m->conf.redis) { } else if (m->conf.redis) {
redis_update(c, m->conf.redis, MASTER_REDIS_ROLE);
redis_update(c, m->conf.redis);
} }
ilog(LOG_INFO, "Returning to SIP proxy: "STR_FORMAT"", STR_FMT0(ret)); ilog(LOG_INFO, "Returning to SIP proxy: "STR_FORMAT"", STR_FMT0(ret));
@ -720,9 +720,9 @@ static const char *call_offer_answer_ng(bencode_item_t *input, struct callmaster
rwlock_unlock_w(&call->master_lock); rwlock_unlock_w(&call->master_lock);
if (m->conf.redis_write) { if (m->conf.redis_write) {
redis_update(call, m->conf.redis_write, ANY_REDIS_ROLE);
redis_update(call, m->conf.redis_write);
} else if (m->conf.redis) { } else if (m->conf.redis) {
redis_update(call, m->conf.redis, MASTER_REDIS_ROLE);
redis_update(call, m->conf.redis);
} }
obj_put(call); obj_put(call);


+ 2
- 2
daemon/main.c View File

@ -587,10 +587,10 @@ no_kernel:
// restore // restore
if (mc.redis_read) { if (mc.redis_read) {
if (redis_restore(ctx->m, mc.redis_read, ANY_REDIS_ROLE))
if (redis_restore(ctx->m, mc.redis_read))
die("Refusing to continue without working Redis read database"); die("Refusing to continue without working Redis read database");
} else if (mc.redis) { } else if (mc.redis) {
if (redis_restore(ctx->m, mc.redis, MASTER_REDIS_ROLE))
if (redis_restore(ctx->m, mc.redis))
die("Refusing to continue without working Redis database"); die("Refusing to continue without working Redis database");
} }


+ 2
- 2
daemon/media_socket.c View File

@ -1397,9 +1397,9 @@ out:
if (ca && update) { if (ca && update) {
if (ca->callmaster->conf.redis_write) { if (ca->callmaster->conf.redis_write) {
redis_update(ca, ca->callmaster->conf.redis, ANY_REDIS_ROLE);
redis_update(ca, ca->callmaster->conf.redis);
} else if (ca->callmaster->conf.redis) { } else if (ca->callmaster->conf.redis) {
redis_update(ca, ca->callmaster->conf.redis, MASTER_REDIS_ROLE);
redis_update(ca, ca->callmaster->conf.redis);
} }
} }
done: done:


+ 19
- 18
daemon/redis.c View File

@ -124,7 +124,7 @@ static void redis_consume(struct redis *r) {
/* called with r->lock held if necessary */ /* called with r->lock held if necessary */
static int redis_connect(struct redis *r, int wait, int role) {
static int redis_connect(struct redis *r, int wait) {
struct timeval tv; struct timeval tv;
redisReply *rp; redisReply *rp;
char *s; char *s;
@ -161,18 +161,18 @@ static int redis_connect(struct redis *r, int wait, int role) {
} }
if (!memcmp(s, "role:master", 9)) { if (!memcmp(s, "role:master", 9)) {
if (role == MASTER_REDIS_ROLE || role == ANY_REDIS_ROLE) {
if (r->role == MASTER_REDIS_ROLE || r->role == ANY_REDIS_ROLE) {
ilog(LOG_INFO, "Connected to Redis in master mode"); ilog(LOG_INFO, "Connected to Redis in master mode");
goto done; goto done;
} else if (role == SLAVE_REDIS_ROLE) {
} else if (r->role == SLAVE_REDIS_ROLE) {
ilog(LOG_INFO, "Connected to Redis in master mode, but wanted mode is slave; retrying..."); ilog(LOG_INFO, "Connected to Redis in master mode, but wanted mode is slave; retrying...");
goto next; goto next;
} }
} else if (!memcmp(s, "role:slave", 8)) { } else if (!memcmp(s, "role:slave", 8)) {
if (role == SLAVE_REDIS_ROLE || role == ANY_REDIS_ROLE) {
if (r->role == SLAVE_REDIS_ROLE || r->role == ANY_REDIS_ROLE) {
ilog(LOG_INFO, "Connected to Redis in slave mode"); ilog(LOG_INFO, "Connected to Redis in slave mode");
goto done; goto done;
} else if (role == MASTER_REDIS_ROLE) {
} else if (r->role == MASTER_REDIS_ROLE) {
ilog(LOG_INFO, "Connected to Redis in slave mode, but wanted mode is master; retrying..."); ilog(LOG_INFO, "Connected to Redis in slave mode, but wanted mode is master; retrying...");
goto next; goto next;
} }
@ -206,7 +206,7 @@ err:
struct redis *redis_new(const endpoint_t *ep, int db, int role) {
struct redis *redis_new(const endpoint_t *ep, int db, enum redis_role role) {
struct redis *r; struct redis *r;
r = g_slice_alloc0(sizeof(*r)); r = g_slice_alloc0(sizeof(*r));
@ -214,9 +214,10 @@ struct redis *redis_new(const endpoint_t *ep, int db, int role) {
r->endpoint = *ep; r->endpoint = *ep;
sockaddr_print(&ep->address, r->host, sizeof(r->host)); sockaddr_print(&ep->address, r->host, sizeof(r->host));
r->db = db; r->db = db;
r->role = role;
mutex_init(&r->lock); mutex_init(&r->lock);
if (redis_connect(r, 10, role))
if (redis_connect(r, 10))
goto err; goto err;
return r; return r;
@ -239,11 +240,11 @@ static void redis_close(struct redis *r) {
/* called with r->lock held if necessary */ /* called with r->lock held if necessary */
static void redis_check_conn(struct redis *r, int role) {
static void redis_check_conn(struct redis *r) {
if (redisCommandNR(r->ctx, "PING") == 0) if (redisCommandNR(r->ctx, "PING") == 0)
return; return;
rlog(LOG_INFO, "Lost connection to Redis"); rlog(LOG_INFO, "Lost connection to Redis");
if (redis_connect(r, 1, role))
if (redis_connect(r, 1))
abort(); abort();
} }
@ -1048,7 +1049,7 @@ static void restore_thread(void *call_p, void *ctx_p) {
mutex_unlock(&ctx->r_m); mutex_unlock(&ctx->r_m);
} }
int redis_restore(struct callmaster *m, struct redis *r, int role) {
int redis_restore(struct callmaster *m, struct redis *r) {
redisReply *calls, *call; redisReply *calls, *call;
int i, ret = -1; int i, ret = -1;
GThreadPool *gtp; GThreadPool *gtp;
@ -1060,7 +1061,7 @@ int redis_restore(struct callmaster *m, struct redis *r, int role) {
log_level |= LOG_FLAG_RESTORE; log_level |= LOG_FLAG_RESTORE;
rlog(LOG_DEBUG, "Restoring calls from Redis..."); rlog(LOG_DEBUG, "Restoring calls from Redis...");
redis_check_conn(r, role);
redis_check_conn(r);
calls = redis_get(r, REDIS_REPLY_ARRAY, "SMEMBERS calls"); calls = redis_get(r, REDIS_REPLY_ARRAY, "SMEMBERS calls");
@ -1073,7 +1074,7 @@ int redis_restore(struct callmaster *m, struct redis *r, int role) {
mutex_init(&ctx.r_m); mutex_init(&ctx.r_m);
g_queue_init(&ctx.r_q); g_queue_init(&ctx.r_q);
for (i = 0; i < RESTORE_NUM_THREADS; i++) for (i = 0; i < RESTORE_NUM_THREADS; i++)
g_queue_push_tail(&ctx.r_q, redis_new(&r->endpoint, r->db, role));
g_queue_push_tail(&ctx.r_q, redis_new(&r->endpoint, r->db, r->role));
gtp = g_thread_pool_new(restore_thread, &ctx, RESTORE_NUM_THREADS, TRUE, NULL); gtp = g_thread_pool_new(restore_thread, &ctx, RESTORE_NUM_THREADS, TRUE, NULL);
for (i = 0; i < calls->elements; i++) { for (i = 0; i < calls->elements; i++) {
@ -1185,7 +1186,7 @@ static void redis_update_dtls_fingerprint(struct redis *r, const char *pref, con
*/ */
/* must be called lock-free */ /* must be called lock-free */
void redis_update(struct call *c, struct redis *r, int role) {
void redis_update(struct call *c, struct redis *r) {
GList *l, *n, *k, *m; GList *l, *n, *k, *m;
struct call_monologue *ml, *ml2; struct call_monologue *ml, *ml2;
struct call_media *media; struct call_media *media;
@ -1199,7 +1200,7 @@ void redis_update(struct call *c, struct redis *r, int role) {
return; return;
mutex_lock(&r->lock); mutex_lock(&r->lock);
redis_check_conn(r, role);
redis_check_conn(r);
rwlock_lock_r(&c->master_lock); rwlock_lock_r(&c->master_lock);
@ -1444,12 +1445,12 @@ void redis_update(struct call *c, struct redis *r, int role) {
/* must be called lock-free */ /* must be called lock-free */
void redis_delete(struct call *c, struct redis *r, int role) {
void redis_delete(struct call *c, struct redis *r) {
if (!r) if (!r)
return; return;
mutex_lock(&r->lock); mutex_lock(&r->lock);
redis_check_conn(r, role);
redis_check_conn(r);
rwlock_lock_r(&c->master_lock); rwlock_lock_r(&c->master_lock);
redis_delete_call(c, r); redis_delete_call(c, r);
@ -1462,12 +1463,12 @@ void redis_delete(struct call *c, struct redis *r, int role) {
void redis_wipe(struct redis *r, int role) {
void redis_wipe(struct redis *r) {
if (!r) if (!r)
return; return;
mutex_lock(&r->lock); mutex_lock(&r->lock);
redis_check_conn(r, role);
redis_check_conn(r);
redisCommandNR(r->ctx, "DEL calls"); redisCommandNR(r->ctx, "DEL calls");
mutex_unlock(&r->lock); mutex_unlock(&r->lock);
} }

+ 11
- 8
daemon/redis.h View File

@ -14,9 +14,11 @@
#include <hiredis/hiredis.h> #include <hiredis/hiredis.h>
#define MASTER_REDIS_ROLE 0
#define SLAVE_REDIS_ROLE 1
#define ANY_REDIS_ROLE 2
enum redis_role {
MASTER_REDIS_ROLE = 0,
SLAVE_REDIS_ROLE = 1,
ANY_REDIS_ROLE = 2,
};
struct callmaster; struct callmaster;
struct call; struct call;
@ -26,6 +28,7 @@ struct call;
struct redis { struct redis {
endpoint_t endpoint; endpoint_t endpoint;
char host[64]; char host[64];
enum redis_role role;
redisContext *ctx; redisContext *ctx;
int db; int db;
@ -75,11 +78,11 @@ INLINE gboolean g_hash_table_insert_check(GHashTable *h, gpointer k, gpointer v)
struct redis *redis_new(const endpoint_t *, int, int);
int redis_restore(struct callmaster *, struct redis *, int);
void redis_update(struct call *, struct redis *, int);
void redis_delete(struct call *, struct redis *, int);
void redis_wipe(struct redis *, int);
struct redis *redis_new(const endpoint_t *, int, enum redis_role);
int redis_restore(struct callmaster *, struct redis *);
void redis_update(struct call *, struct redis *);
void redis_delete(struct call *, struct redis *);
void redis_wipe(struct redis *);


Write Preview
Loading…
Cancel
Save