#include "announce.h"
#include "util.h"
-#define FLX_ANNOUNCEMENT_JITTER_MSEC 0
+#define FLX_ANNOUNCEMENT_JITTER_MSEC 250
+#define FLX_PROBE_JITTER_MSEC 250
+#define FLX_PROBE_INTERVAL_MSEC 250
static void remove_announcement(flxServer *s, flxAnnouncement *a) {
g_assert(s);
g_assert(a);
- flx_time_event_queue_remove(s->time_event_queue, a->time_event);
+ if (a->time_event)
+ flx_time_event_queue_remove(s->time_event_queue, a->time_event);
FLX_LLIST_REMOVE(flxAnnouncement, by_interface, a->interface->announcements, a);
FLX_LLIST_REMOVE(flxAnnouncement, by_entry, a->entry->announcements, a);
static void elapse_announce(flxTimeEvent *e, void *userdata);
-static void send_packet(flxAnnouncement *a) {
- GTimeVal tv;
+static void set_timeout(flxAnnouncement *a, const GTimeVal *tv) {
g_assert(a);
-/* g_message("%i -- %u", a->state, a->n_iteration); */
+ if (!tv) {
+ if (a->time_event) {
+ flx_time_event_queue_remove(a->server->time_event_queue, a->time_event);
+ a->time_event = NULL;
+ }
+ } else {
+
+ if (a->time_event)
+ flx_time_event_queue_update(a->server->time_event_queue, a->time_event, tv);
+ else
+ a->time_event = flx_time_event_queue_add(a->server->time_event_queue, tv, elapse_announce, a);
+ }
+}
+
+static void next_state(flxAnnouncement *a);
+
+void flx_entry_group_check_probed(flxEntryGroup *g, gboolean immediately) {
+ flxEntry *e;
+ g_assert(g);
+ g_assert(!g->dead);
+
+ /* Check whether all group members have been probed */
- if (a->state == FLX_PROBING && a->n_iteration >= 1) {
- flx_interface_post_probe(a->interface, a->entry->record, FALSE);
- } else if (a->state == FLX_ANNOUNCING && a->n_iteration >= 1)
- flx_interface_post_response(a->interface, NULL, a->entry->record, a->entry->flags & FLX_SERVER_ENTRY_UNIQUE, TRUE);
+ if (g->status != FLX_ENTRY_GROUP_REGISTERING || g->n_probing > 0)
+ return;
- a->n_iteration++;
+ flx_entry_group_run_callback(g, g->status = FLX_ENTRY_GROUP_ESTABLISHED);
- if (a->state == FLX_PROBING) {
+ if (g->dead)
+ return;
+
+ for (e = g->entries; e; e = e->entries_next) {
+ flxAnnouncement *a;
+
+ for (a = e->announcements; a; a = a->by_entry_next) {
- if (a->n_iteration == 1)
- flx_elapse_time(&tv, 0, 250);
- else
- flx_elapse_time(&tv, 250, 0);
+ if (a->state != FLX_WAITING)
+ continue;
+
+ a->state = FLX_ANNOUNCING;
+
+ if (immediately) {
+ /* Shortcut */
+
+ a->n_iteration = 1;
+ next_state(a);
+ } else {
+ GTimeVal tv;
+ a->n_iteration = 0;
+ flx_elapse_time(&tv, 0, FLX_ANNOUNCEMENT_JITTER_MSEC);
+ set_timeout(a, &tv);
+ }
+ }
+ }
+}
+
+static void next_state(flxAnnouncement *a) {
+ g_assert(a);
+
+ g_message("%i -- %u", a->state, a->n_iteration);
+
+ if (a->state == FLX_WAITING) {
+
+ g_assert(a->entry->group);
+
+ flx_entry_group_check_probed(a->entry->group, TRUE);
+
+ } else if (a->state == FLX_PROBING) {
- /* Probing done */
if (a->n_iteration >= 4) {
+ /* Probing done */
+
gchar *t;
+
g_message("Enough probes for record [%s]", t = flx_record_to_string(a->entry->record));
g_free(t);
- a->state = FLX_ANNOUNCING;
- a->n_iteration = 1;
+
+ if (a->entry->group) {
+ g_assert(a->entry->group->n_probing);
+ a->entry->group->n_probing--;
+ }
+
+ if (a->entry->group && a->entry->group->status == FLX_ENTRY_GROUP_REGISTERING)
+ a->state = FLX_WAITING;
+ else {
+ a->state = FLX_ANNOUNCING;
+ a->n_iteration = 1;
+ }
+
+ set_timeout(a, NULL);
+ next_state(a);
+ } else {
+ GTimeVal tv;
+
+ flx_interface_post_probe(a->interface, a->entry->record, FALSE);
+
+ flx_elapse_time(&tv, FLX_PROBE_INTERVAL_MSEC, 0);
+ set_timeout(a, &tv);
+
+ a->n_iteration++;
}
-
+
} else if (a->state == FLX_ANNOUNCING) {
- flx_elapse_time(&tv, a->sec_delay*1000, FLX_ANNOUNCEMENT_JITTER_MSEC);
-
- if (a->n_iteration < 10)
- a->sec_delay *= 2;
+ flx_interface_post_response(a->interface, NULL, a->entry->record, a->entry->flags & FLX_ENTRY_UNIQUE, FALSE);
- /* Announcing done */
- if (a->n_iteration >= 4) {
+ if (++a->n_iteration >= 4) {
gchar *t;
+ /* Announcing done */
+
g_message("Enough announcements for record [%s]", t = flx_record_to_string(a->entry->record));
g_free(t);
- remove_announcement(a->server, a);
- return;
+
+ a->state = FLX_ESTABLISHED;
+
+ set_timeout(a, NULL);
+ } else {
+ GTimeVal tv;
+ flx_elapse_time(&tv, a->sec_delay*1000, FLX_ANNOUNCEMENT_JITTER_MSEC);
+
+ if (a->n_iteration < 10)
+ a->sec_delay *= 2;
+
+ set_timeout(a, &tv);
}
}
-
- if (a->time_event)
- flx_time_event_queue_update(a->server->time_event_queue, a->time_event, &tv);
- else
- a->time_event = flx_time_event_queue_add(a->server->time_event_queue, &tv, elapse_announce, a);
}
static void elapse_announce(flxTimeEvent *e, void *userdata) {
g_assert(e);
- send_packet(userdata);
+ next_state(userdata);
}
-static flxAnnouncement *get_announcement(flxServer *s, flxServerEntry *e, flxInterface *i) {
+flxAnnouncement *flx_get_announcement(flxServer *s, flxEntry *e, flxInterface *i) {
flxAnnouncement *a;
g_assert(s);
return NULL;
}
-static void new_announcement(flxServer *s, flxInterface *i, flxServerEntry *e) {
+static void new_announcement(flxServer *s, flxInterface *i, flxEntry *e) {
flxAnnouncement *a;
GTimeVal tv;
gchar *t;
g_assert(s);
g_assert(i);
g_assert(e);
+ g_assert(!e->dead);
/* g_message("NEW ANNOUNCEMENT: %s.%i [%s]", i->hardware->name, i->protocol, t = flx_record_to_string(e->record)); */
/* g_free(t); */
- if (!flx_interface_match(i, e->interface, e->protocol) || !i->announcing || e->flags & FLX_SERVER_ENTRY_NOANNOUNCE)
+ if (!flx_interface_match(i, e->interface, e->protocol) || !i->announcing || !flx_entry_commited(e))
return;
/* We don't want duplicate announcements */
- if (get_announcement(s, e, i))
+ if (flx_get_announcement(s, e, i))
return;
- g_message("New announcement on interface %s.%i for entry [%s]", i->hardware->name, i->protocol, t = flx_record_to_string(e->record));
- g_free(t);
-
a = g_new(flxAnnouncement, 1);
a->server = s;
a->interface = i;
a->entry = e;
- a->state = (e->flags & FLX_SERVER_ENTRY_UNIQUE) && !(e->flags & FLX_SERVER_ENTRY_NOPROBE) ? FLX_PROBING : FLX_ANNOUNCING;
- a->n_iteration = 0;
+ if ((e->flags & FLX_ENTRY_UNIQUE) && !(e->flags & FLX_ENTRY_NOPROBE))
+ a->state = FLX_PROBING;
+ else if (!(e->flags & FLX_ENTRY_NOANNOUNCE)) {
+
+ if (!e->group || e->group->status == FLX_ENTRY_GROUP_ESTABLISHED)
+ a->state = FLX_ANNOUNCING;
+ else
+ a->state = FLX_WAITING;
+
+ } else
+ a->state = FLX_ESTABLISHED;
+
+
+ g_message("New announcement on interface %s.%i for entry [%s] state=%i", i->hardware->name, i->protocol, t = flx_record_to_string(e->record), a->state);
+ g_free(t);
+
+ a->n_iteration = 1;
a->sec_delay = 1;
a->time_event = NULL;
+
+ if (a->state == FLX_PROBING)
+ if (e->group)
+ e->group->n_probing++;
FLX_LLIST_PREPEND(flxAnnouncement, by_interface, i->announcements, a);
FLX_LLIST_PREPEND(flxAnnouncement, by_entry, e->announcements, a);
- send_packet(a);
+ if (a->state == FLX_PROBING) {
+ flx_elapse_time(&tv, 0, FLX_PROBE_JITTER_MSEC);
+ set_timeout(a, &tv);
+ } else if (a->state == FLX_ANNOUNCING) {
+ flx_elapse_time(&tv, 0, FLX_ANNOUNCEMENT_JITTER_MSEC);
+ set_timeout(a, &tv);
+ }
}
void flx_announce_interface(flxServer *s, flxInterface *i) {
- flxServerEntry *e;
+ flxEntry *e;
g_assert(s);
g_assert(i);
if (!i->announcing)
return;
-/* g_message("ANNOUNCE INTERFACE"); */
-
- for (e = s->entries; e; e = e->entry_next)
- new_announcement(s, i, e);
+ for (e = s->entries; e; e = e->entries_next)
+ if (!e->dead)
+ new_announcement(s, i, e);
}
static void announce_walk_callback(flxInterfaceMonitor *m, flxInterface *i, gpointer userdata) {
- flxServerEntry *e = userdata;
+ flxEntry *e = userdata;
g_assert(m);
g_assert(i);
g_assert(e);
+ g_assert(!e->dead);
new_announcement(m->server, i, e);
}
-void flx_announce_entry(flxServer *s, flxServerEntry *e) {
+void flx_announce_entry(flxServer *s, flxEntry *e) {
g_assert(s);
g_assert(e);
-
-/* g_message("ANNOUNCE ENTRY"); */
+ g_assert(!e->dead);
flx_interface_monitor_walk(s->monitor, e->interface, e->protocol, announce_walk_callback, e);
}
-gboolean flx_entry_established(flxServer *s, flxServerEntry *e, flxInterface *i) {
+void flx_announce_group(flxServer *s, flxEntryGroup *g) {
+ flxEntry *e;
+
+ g_assert(s);
+ g_assert(g);
+
+ for (e = g->entries; e; e = e->by_group_next)
+ if (!e->dead)
+ flx_announce_entry(s, e);
+}
+
+gboolean flx_entry_registered(flxServer *s, flxEntry *e, flxInterface *i) {
flxAnnouncement *a;
g_assert(s);
g_assert(e);
g_assert(i);
+ g_assert(!e->dead);
+
+ if (!(a = flx_get_announcement(s, e, i)))
+ return FALSE;
+
+ return a->state == FLX_ANNOUNCING || a->state == FLX_ESTABLISHED;
+}
- if (!(e->flags & FLX_SERVER_ENTRY_UNIQUE) || (e->flags & FLX_SERVER_ENTRY_NOPROBE))
- return TRUE;
+gboolean flx_entry_registering(flxServer *s, flxEntry *e, flxInterface *i) {
+ flxAnnouncement *a;
- if ((a = get_announcement(s, e, i)))
- if (a->state == FLX_PROBING)
- return FALSE;
+ g_assert(s);
+ g_assert(e);
+ g_assert(i);
+ g_assert(!e->dead);
- return TRUE;
+ if (!(a = flx_get_announcement(s, e, i)))
+ return FALSE;
+
+ return a->state == FLX_PROBING || a->state == FLX_WAITING;
}
-
static flxRecord *make_goodbye_record(flxRecord *r) {
gchar *t;
flxRecord *g;
return g;
}
-
static void send_goodbye_callback(flxInterfaceMonitor *m, flxInterface *i, gpointer userdata) {
- flxServerEntry *e = userdata;
+ flxEntry *e = userdata;
flxRecord *g;
g_assert(m);
g_assert(i);
g_assert(e);
+ g_assert(!e->dead);
if (!flx_interface_match(i, e->interface, e->protocol))
return;
- if (e->flags & FLX_SERVER_ENTRY_NOANNOUNCE)
+ if (e->flags & FLX_ENTRY_NOANNOUNCE)
return;
- if (!flx_entry_established(m->server, e, i))
+ if (!flx_entry_registered(m->server, e, i))
return;
g = make_goodbye_record(e->record);
- flx_interface_post_response(i, NULL, g, e->flags & FLX_SERVER_ENTRY_UNIQUE, TRUE);
+ flx_interface_post_response(i, NULL, g, e->flags & FLX_ENTRY_UNIQUE, TRUE);
flx_record_unref(g);
}
g_message("goodbye interface: %s.%u", i->hardware->name, i->protocol);
if (goodbye && flx_interface_relevant(i)) {
- flxServerEntry *e;
+ flxEntry *e;
- for (e = s->entries; e; e = e->entry_next)
- send_goodbye_callback(s->monitor, i, e);
+ for (e = s->entries; e; e = e->entries_next)
+ if (!e->dead)
+ send_goodbye_callback(s->monitor, i, e);
}
while (i->announcements)
}
-void flx_goodbye_entry(flxServer *s, flxServerEntry *e, gboolean goodbye) {
+void flx_goodbye_entry(flxServer *s, flxEntry *e, gboolean goodbye) {
g_assert(s);
g_assert(e);
-
+ g_assert(!e->dead);
+
g_message("goodbye entry: %p", e);
if (goodbye)
}
void flx_goodbye_all(flxServer *s, gboolean goodbye) {
- flxServerEntry *e;
+ flxEntry *e;
g_assert(s);
g_message("goodbye all");
- for (e = s->entries; e; e = e->entry_next)
- flx_goodbye_entry(s, e, goodbye);
+ for (e = s->entries; e; e = e->entries_next)
+ if (!e->dead)
+ flx_goodbye_entry(s, e, goodbye);
g_message("goodbye all done");
typedef enum {
FLX_PROBING,
+ FLX_WAITING, /* wait for other records in group */
FLX_ANNOUNCING,
+ FLX_ESTABLISHED
} flxAnnouncementState;
struct _flxAnnouncement {
flxServer *server;
flxInterface *interface;
- flxServerEntry *entry;
+ flxEntry *entry;
flxTimeEvent *time_event;
};
void flx_announce_interface(flxServer *s, flxInterface *i);
-void flx_announce_entry(flxServer *s, flxServerEntry *e);
+void flx_announce_entry(flxServer *s, flxEntry *e);
+void flx_announce_group(flxServer *s, flxEntryGroup *g);
-gboolean flx_entry_established(flxServer *s, flxServerEntry *e, flxInterface *i);
+void flx_entry_group_check_probed(flxEntryGroup *g, gboolean immediately);
+
+gboolean flx_entry_registered(flxServer *s, flxEntry *e, flxInterface *i);
+gboolean flx_entry_registering(flxServer *s, flxEntry *e, flxInterface *i);
void flx_goodbye_interface(flxServer *s, flxInterface *i, gboolean send);
-void flx_goodbye_entry(flxServer *s, flxServerEntry *e, gboolean send);
+void flx_goodbye_entry(flxServer *s, flxEntry *e, gboolean send);
void flx_goodbye_all(flxServer *s, gboolean send);
+flxAnnouncement *flx_get_announcement(flxServer *s, flxEntry *e, flxInterface *i);
+
#endif
return g_ntohs(((guint16*) FLX_DNS_PACKET_DATA(p))[index]);
}
-/* Read the first label from string dest, unescape "\" and append it to *name */
-static gchar *unescape_label(gchar *dest, guint size, const gchar **name) {
+/* Read the first label from string *name, unescape "\" and write it to dest */
+gchar *flx_unescape_label(gchar *dest, guint size, const gchar **name) {
guint i = 0;
gchar *d;
guint n;
guint8* prev;
const gchar *pname;
- char label[64];
+ gchar label[64];
/* Check whether we can compress this name. */
pname = name;
- if (!(unescape_label(label, sizeof(label), &name)))
+ if (!(flx_unescape_label(label, sizeof(label), &name)))
goto fail;
if (!(d = flx_dns_packet_append_string(p, label)))
((guint16) (rd & 15)))
+gchar *flx_unescape_label(gchar *dest, guint size, const gchar **name);
+
#endif
#include <glib.h>
typedef struct _flxServer flxServer;
+typedef struct _flxEntry flxEntry;
+typedef struct _flxEntryGroup flxEntryGroup;
#include "address.h"
#include "rr.h"
typedef enum {
- FLX_SERVER_ENTRY_NULL = 0,
- FLX_SERVER_ENTRY_UNIQUE = 1,
- FLX_SERVER_ENTRY_NOPROBE = 2,
- FLX_SERVER_ENTRY_NOANNOUNCE = 4
-} flxServerEntryFlags;
+ FLX_ENTRY_NULL = 0,
+ FLX_ENTRY_UNIQUE = 1,
+ FLX_ENTRY_NOPROBE = 2,
+ FLX_ENTRY_NOANNOUNCE = 4
+} flxEntryFlags;
+
+typedef enum {
+ FLX_ENTRY_GROUP_UNCOMMITED,
+ FLX_ENTRY_GROUP_REGISTERING,
+ FLX_ENTRY_GROUP_ESTABLISHED,
+ FLX_ENTRY_GROUP_COLLISION
+} flxEntryGroupStatus;
+
+typedef void (*flxEntryGroupCallback) (flxServer *s, flxEntryGroup *g, flxEntryGroupStatus status, gpointer userdata);
flxServer *flx_server_new(GMainContext *c);
void flx_server_free(flxServer* s);
-gint flx_server_get_next_id(flxServer *s);
+const flxRecord *flx_server_iterate(flxServer *s, flxEntryGroup *g, void **state);
+void flx_server_dump(flxServer *s, FILE *f);
+
+flxEntryGroup *flx_entry_group_new(flxServer *s, flxEntryGroupCallback callback, gpointer userdata);
+void flx_entry_group_free(flxEntryGroup *g);
+void flx_entry_group_commit(flxEntryGroup *g);
+flxEntryGroupStatus flx_entry_group_get_status(flxEntryGroup *g);
void flx_server_add(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
- flxServerEntryFlags flags,
+ flxEntryFlags flags,
flxRecord *r);
void flx_server_add_ptr(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
- flxServerEntryFlags flags,
+ flxEntryFlags flags,
const gchar *name,
const gchar *dest);
void flx_server_add_address(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
- flxServerEntryFlags flags,
+ flxEntryFlags flags,
const gchar *name,
flxAddress *a);
void flx_server_add_text(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
- flxServerEntryFlags flags,
+ flxEntryFlags flags,
const gchar *name,
... /* text records, terminated by NULL */);
void flx_server_add_text_va(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
- flxServerEntryFlags flags,
+ flxEntryFlags flags,
const gchar *name,
va_list va);
void flx_server_add_service(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
const gchar *type,
void flx_server_add_service_va(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
const gchar *type,
guint16 port,
va_list va);
+typedef enum {
+ FLX_SUBSCRIPTION_NEW,
+ FLX_SUBSCRIPTION_REMOVE,
+ FLX_SUBSCRIPTION_CHANGE
+} flxSubscriptionEvent;
-void flx_server_remove(flxServer *s, gint id);
-
-void flx_server_post_query(flxServer *s, gint interface, guchar protocol, flxKey *key);
-void flx_server_post_response(flxServer *s, gint interface, guchar protocol, flxRecord *record, gboolean flush_cache);
+typedef struct _flxSubscription flxSubscription;
-const flxRecord *flx_server_iterate(flxServer *s, gint id, void **state);
+typedef void (*flxSubscriptionCallback)(flxSubscription *s, flxRecord *record, gint interface, guchar protocol, flxSubscriptionEvent event, gpointer userdata);
-void flx_server_dump(flxServer *s, FILE *f);
+flxSubscription *flx_subscription_new(flxServer *s, flxKey *key, gint interface, guchar protocol, flxSubscriptionCallback callback, gpointer userdata);
+void flx_subscription_free(flxSubscription *s);
#endif
g_assert(a);
if (!flx_interface_address_relevant(a) || remove) {
- if (a->rr_id >= 0) {
- flx_server_remove(m->server, a->rr_id);
- a->rr_id = -1;
+ if (a->entry_group) {
+ flx_entry_group_free(a->entry_group);
+ a->entry_group = NULL;
}
} else {
- if (a->rr_id < 0) {
- a->rr_id = flx_server_get_next_id(m->server);
- flx_server_add_address(m->server, a->rr_id, a->interface->hardware->index, AF_UNSPEC, 0, NULL, &a->address);
+ if (!a->entry_group) {
+/* a->entry_group = flx_entry_group_new(m->server, NULL, NULL); */
+/* flx_server_add_address(m->server, a->entry_group, a->interface->hardware->index, AF_UNSPEC, 0, NULL, &a->address); */
+/* flx_entry_group_commit(a->entry_group); */
}
}
}
g_assert(a->interface);
FLX_LLIST_REMOVE(flxInterfaceAddress, address, a->interface->addresses, a);
- flx_server_remove(m->server, a->rr_id);
+
+ if (a->entry_group)
+ flx_entry_group_free(a->entry_group);
g_free(a);
}
addr->monitor = m;
addr->address = raddr;
addr->interface = i;
- addr->rr_id = -1;
+ addr->entry_group = NULL;
FLX_LLIST_PREPEND(flxInterfaceAddress, address, i->addresses, addr);
}
m = g_new0(flxInterfaceMonitor, 1);
m->server = s;
- if (!(m->netlink = flx_netlink_new(s->context, RTMGRP_LINK|RTMGRP_IPV4_IFADDR|RTMGRP_IPV6_IFADDR, callback, m)))
+ if (!(m->netlink = flx_netlink_new(s->context, G_PRIORITY_DEFAULT-10, RTMGRP_LINK|RTMGRP_IPV4_IFADDR|RTMGRP_IPV6_IFADDR, callback, m)))
goto fail;
m->hash_table = g_hash_table_new(g_int_hash, g_int_equal);
goto fail;
m->list = LIST_IFACE;
-
+
return m;
fail:
return NULL;
}
+void flx_interface_monitor_sync(flxInterfaceMonitor *m) {
+ g_assert(m);
+
+ while (m->list != LIST_DONE) {
+ if (!flx_netlink_work(m->netlink, TRUE))
+ break;
+ }
+}
+
void flx_interface_monitor_free(flxInterfaceMonitor *m) {
g_assert(m);
guchar scope;
flxAddress address;
- gint rr_id;
+ flxEntryGroup *entry_group;
flxInterface *interface;
};
flxInterfaceMonitor *flx_interface_monitor_new(flxServer *server);
void flx_interface_monitor_free(flxInterfaceMonitor *m);
+void flx_interface_monitor_sync(flxInterfaceMonitor *m);
+
flxInterface* flx_interface_monitor_get_interface(flxInterfaceMonitor *m, gint index, guchar protocol);
flxHwInterface* flx_interface_monitor_get_hw_interface(flxInterfaceMonitor *m, gint index);
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
+#include <stdlib.h>
#include "flx.h"
-#include "server.h"
-#include "subscribe.h"
static gboolean quit_timeout(gpointer data) {
g_main_loop_quit(data);
return FALSE;
}
-static gboolean send_timeout(gpointer data) {
- flxServer *flx = data;
- flxKey *k;
-
-/* k = flx_key_new("ecstasy.local.", FLX_DNS_CLASS_IN, FLX_DNS_TYPE_TXT); */
-/* flx_server_post_query(flx, 0, AF_UNSPEC, k); */
-/* flx_key_unref(k); */
-
-/* k = flx_key_new("ecstasy.local.", FLX_DNS_CLASS_IN, FLX_DNS_TYPE_A); */
-/* flx_server_post_query(flx, 0, AF_INET, k); */
-/* flx_key_unref(k); */
-
- return FALSE;
-}
-
static gboolean dump_timeout(gpointer data) {
flxServer *flx = data;
flx_server_dump(flx, stdout);
g_free(t);
}
+static void entry_group_callback(flxServer *s, flxEntryGroup *g, flxEntryGroupStatus status, gpointer userdata) {
+ g_message("entry group state: %i", status);
+}
int main(int argc, char *argv[]) {
flxServer *flx;
GMainLoop *loop = NULL;
flxSubscription *s;
flxKey *k;
+ flxEntryGroup *g;
flx = flx_server_new(NULL);
- flx_server_add_text(flx, 0, 0, AF_UNSPEC, FLX_SERVER_ENTRY_UNIQUE, NULL, "hallo", NULL);
- flx_server_add_service(flx, 0, 0, AF_UNSPEC, "_http._tcp", "gurke", NULL, NULL, 80, "foo", NULL);
+ g = flx_entry_group_new(flx, entry_group_callback, NULL);
+
+/* flx_server_add_text(flx, g, 0, AF_UNSPEC, FLX_ENTRY_UNIQUE, NULL, "hallo", NULL); */
+ flx_server_add_service(flx, g, 0, AF_UNSPEC, "_http._tcp", "gurke", NULL, NULL, 80, "foo", NULL);
+
+ flx_entry_group_commit(g);
+ flx_server_dump(flx, stdout);
+
+
/* k = flx_key_new("ecstasy.local.", FLX_DNS_CLASS_IN, FLX_DNS_TYPE_ANY); */
/* s = flx_subscription_new(flx, k, 0, AF_UNSPEC, subscription, NULL); */
/* flx_key_unref(k); */
loop = g_main_loop_new(NULL, FALSE);
- g_timeout_add(1000*30, quit_timeout, loop);
- g_timeout_add(1000, send_timeout, flx);
g_timeout_add(1000*20, dump_timeout, flx);
+ g_timeout_add(1000*30, quit_timeout, loop);
g_main_loop_run(loop);
-
g_main_loop_unref(loop);
/* flx_subscription_free(s); */
+/* flx_entry_group_free(g); */
flx_server_free(flx);
return 0;
void (*callback) (flxNetlink *nl, struct nlmsghdr *n, gpointer userdata);
gpointer userdata;
};
-static gboolean work(flxNetlink *nl) {
+
+gboolean flx_netlink_work(flxNetlink *nl, gboolean block) {
g_assert(nl);
for (;;) {
ssize_t bytes;
struct nlmsghdr *p = (struct nlmsghdr *) replybuf;
- if ((bytes = recv(nl->fd, replybuf, sizeof(replybuf), MSG_DONTWAIT)) < 0) {
+ if ((bytes = recv(nl->fd, replybuf, sizeof(replybuf), block ? 0 : MSG_DONTWAIT)) < 0) {
if (errno == EAGAIN || errno == EINTR)
break;
if (nl->callback) {
for (; bytes > 0; p = NLMSG_NEXT(p, bytes)) {
- if (!NLMSG_OK(p, bytes)) {
+ if (!NLMSG_OK(p, (size_t) bytes)) {
g_warning("NETLINK: packet truncated");
return FALSE;
}
nl->callback(nl, p, nl->userdata);
}
}
+
+ if (block)
+ break;
}
return TRUE;
nl = *((flxNetlink**) (((guint8*) source) + sizeof(GSource)));
g_assert(nl);
- return work(nl);
+ return flx_netlink_work(nl, FALSE);
}
-flxNetlink *flx_netlink_new(GMainContext *context, guint32 groups, void (*cb) (flxNetlink *nl, struct nlmsghdr *n, gpointer userdata), gpointer userdata) {
+flxNetlink *flx_netlink_new(GMainContext *context, gint priority, guint32 groups, void (*cb) (flxNetlink *nl, struct nlmsghdr *n, gpointer userdata), gpointer userdata) {
int fd;
struct sockaddr_nl addr;
flxNetlink *nl;
nl->source = g_source_new(&source_funcs, sizeof(GSource) + sizeof(flxNetlink*));
*((flxNetlink**) (((guint8*) nl->source) + sizeof(GSource))) = nl;
+ g_source_set_priority(nl->source, priority);
+
memset(&nl->poll_fd, 0, sizeof(GPollFD));
nl->poll_fd.fd = fd;
nl->poll_fd.events = G_IO_IN|G_IO_ERR|G_IO_HUP;
struct _flxNetlink;
typedef struct _flxNetlink flxNetlink;
-flxNetlink *flx_netlink_new(GMainContext *c, guint32 groups, void (*cb) (flxNetlink *n, struct nlmsghdr *m, gpointer userdata), gpointer userdata);
+flxNetlink *flx_netlink_new(GMainContext *c, gint priority, guint32 groups, void (*cb) (flxNetlink *n, struct nlmsghdr *m, gpointer userdata), gpointer userdata);
void flx_netlink_free(flxNetlink *n);
int flx_netlink_send(flxNetlink *n, struct nlmsghdr *m, guint *ret_seq);
+gboolean flx_netlink_work(flxNetlink *n, gboolean block);
+
#endif
#define FLX_RESPONSE_HISTORY_MSEC 700
#define FLX_RESPONSE_DEFER_MSEC 20
#define FLX_RESPONSE_JITTER_MSEC 100
-#define FLX_PROBE_DEFER_MSEC 100
+#define FLX_PROBE_DEFER_MSEC 20
flxPacketScheduler *flx_packet_scheduler_new(flxServer *server, flxInterface *i) {
flxPacketScheduler *s;
qj->done = 1;
- /* Drop query after some time from history from history */
+ /* Drop query after some time from history */
flx_elapse_time(&tv, FLX_QUERY_HISTORY_MSEC, 0);
flx_time_event_queue_update(s->server->time_event_queue, qj->time_event, &tv);
flx_elapse_time(&tv, immediately ? 0 : FLX_QUERY_DEFER_MSEC, 0);
- for (qj = s->query_jobs; qj; qj = qj->jobs_next)
+ for (qj = s->query_jobs; qj; qj = qj->jobs_next) {
if (flx_key_equal(qj->key, key)) {
break;
}
+ }
+
qj = query_job_new(s, key);
qj->delivery = tv;
qj->time_event = flx_time_event_queue_add(s->server->time_event_queue, &qj->delivery, query_elapse, qj);
}
probe_job_free(s, pj);
+
n ++;
}
flx_elapse_time(&tv, immediately ? 0 : FLX_PROBE_DEFER_MSEC, 0);
- /* No duplication check here... */
/* Create a new job and schedule it */
pj = probe_job_new(s, record);
pj->delivery = tv;
#include "util.h"
#include "rr.h"
+#include "dns.h"
flxKey *flx_key_new(const gchar *name, guint16 class, guint16 type) {
flxKey *k;
return n;
}
+
+static gint lexicographical_memcmp(gconstpointer a, size_t al, gconstpointer b, size_t bl) {
+ size_t c;
+ gint ret;
+
+ g_assert(a);
+ g_assert(b);
+
+ c = al < bl ? al : bl;
+ if ((ret = memcmp(a, b, c)) != 0)
+ return ret;
+
+ if (al == bl)
+ return 0;
+ else
+ return al == c ? 1 : -1;
+}
+
+static gint uint16_cmp(guint16 a, guint16 b) {
+ return a == b ? 0 : (a < b ? a : b);
+}
+
+static gint lexicographical_domain_cmp(const gchar *a, const gchar *b) {
+ g_assert(a);
+ g_assert(b);
+
+
+ for (;;) {
+ gchar t1[64];
+ gchar t2[64];
+ size_t al, bl;
+ gint r;
+
+ if (!a && !b)
+ return 0;
+
+ if (a && !b)
+ return 1;
+
+ if (b && !a)
+ return -1;
+
+ flx_unescape_label(t1, sizeof(t1), &a);
+ flx_unescape_label(t2, sizeof(t2), &b);
+
+ al = strlen(t1);
+ bl = strlen(t2);
+
+ if (al != bl)
+ return al < bl ? -1 : 1;
+
+ if ((r = strcmp(t1, t2)) != 0)
+ return r;
+ }
+}
+
+gint flx_record_lexicographical_compare(flxRecord *a, flxRecord *b) {
+ g_assert(a);
+ g_assert(b);
+
+ if (a->key->class < b->key->class)
+ return -1;
+ else if (a->key->class > b->key->class)
+ return 1;
+
+ if (a->key->type < b->key->type)
+ return -1;
+ else if (a->key->type > b->key->type)
+ return 1;
+
+ switch (a->key->type) {
+
+ case FLX_DNS_TYPE_PTR:
+ case FLX_DNS_TYPE_CNAME:
+ return lexicographical_domain_cmp(a->data.ptr.name, b->data.ptr.name);
+
+ case FLX_DNS_TYPE_SRV: {
+ gint r;
+ if ((r = uint16_cmp(a->data.srv.priority, b->data.srv.priority)) != 0 ||
+ (r = uint16_cmp(a->data.srv.weight, b->data.srv.weight)) != 0 ||
+ (r = uint16_cmp(a->data.srv.port, b->data.srv.port)) != 0)
+ return lexicographical_domain_cmp(a->data.srv.name, b->data.srv.name);
+ }
+
+ case FLX_DNS_TYPE_HINFO: {
+ size_t al = strlen(a->data.hinfo.cpu), bl = strlen(b->data.hinfo.cpu);
+ gint r;
+
+ if (al != bl)
+ return al < bl ? -1 : 1;
+
+ if ((r = strcmp(a->data.hinfo.cpu, b->data.hinfo.cpu)) != 0)
+ return r;
+
+ al = strlen(a->data.hinfo.os), bl = strlen(b->data.hinfo.os);
+
+ if (al != bl)
+ return al < bl ? -1 : 1;
+
+ if ((r = strcmp(a->data.hinfo.os, b->data.hinfo.os)) != 0)
+ return r;
+
+ return 0;
+
+ }
+
+ case FLX_DNS_TYPE_TXT: {
+
+ guint8 *ma, *mb;
+ guint asize, bsize;
+ gint r;
+
+ ma = g_new(guint8, asize = flx_string_list_serialize(a->data.txt.string_list, NULL, 0));
+ mb = g_new(guint8, bsize = flx_string_list_serialize(b->data.txt.string_list, NULL, 0));
+ flx_string_list_serialize(a->data.txt.string_list, ma, asize);
+ flx_string_list_serialize(a->data.txt.string_list, mb, bsize);
+
+ r = lexicographical_memcmp(ma, asize, mb, bsize);
+ g_free(ma);
+ g_free(mb);
+
+ return r;
+ }
+
+ case FLX_DNS_TYPE_A:
+ return memcmp(&a->data.a.address, &b->data.a.address, sizeof(flxIPv4Address));
+
+ case FLX_DNS_TYPE_AAAA:
+ return memcmp(&a->data.aaaa.address, &b->data.aaaa.address, sizeof(flxIPv6Address));
+
+ default:
+ return lexicographical_memcmp(a->data.generic.data, a->data.generic.size,
+ b->data.generic.data, b->data.generic.size);
+ }
+
+}
/* ditto */
guint flx_record_get_estimate_size(flxRecord *r);
+gint flx_record_lexicographical_compare(flxRecord *a, flxRecord *b);
#endif
#include "socket.h"
#include "subscribe.h"
+static void free_entry(flxServer*s, flxEntry *e) {
+ flxEntry *t;
+
+ g_assert(s);
+ g_assert(e);
+
+ flx_goodbye_entry(s, e, TRUE);
+
+ /* Remove from linked list */
+ FLX_LLIST_REMOVE(flxEntry, entries, s->entries, e);
+
+ /* Remove from hash table indexed by name */
+ t = g_hash_table_lookup(s->entries_by_key, e->record->key);
+ FLX_LLIST_REMOVE(flxEntry, by_key, t, e);
+ if (t)
+ g_hash_table_replace(s->entries_by_key, t->record->key, t);
+ else
+ g_hash_table_remove(s->entries_by_key, e->record->key);
+
+ /* Remove from associated group */
+ if (e->group)
+ FLX_LLIST_REMOVE(flxEntry, by_group, e->group->entries, e);
+
+ flx_record_unref(e->record);
+ g_free(e);
+}
+
+static void free_group(flxServer *s, flxEntryGroup *g) {
+ g_assert(s);
+ g_assert(g);
+
+ while (g->entries)
+ free_entry(s, g->entries);
+
+ FLX_LLIST_REMOVE(flxEntryGroup, groups, s->groups, g);
+ g_free(g);
+}
+
+static void cleanup_dead(flxServer *s) {
+ flxEntryGroup *g, *ng;
+ flxEntry *e, *ne;
+ g_assert(s);
+
+
+ if (s->need_group_cleanup) {
+ for (g = s->groups; g; g = ng) {
+ ng = g->groups_next;
+
+ if (g->dead)
+ free_group(s, g);
+ }
+
+ s->need_group_cleanup = FALSE;
+ }
+
+ if (s->need_entry_cleanup) {
+ for (e = s->entries; e; e = ne) {
+ ne = e->entries_next;
+
+ if (e->dead)
+ free_entry(s, e);
+ }
+
+ s->need_entry_cleanup = FALSE;
+ }
+}
+
static void handle_query_key(flxServer *s, flxKey *k, flxInterface *i, const flxAddress *a) {
- flxServerEntry *e;
+ flxEntry *e;
gchar *txt;
g_assert(s);
/* Handle ANY query */
- for (e = s->entries; e; e = e->entry_next)
- if (flx_key_pattern_match(k, e->record->key))
- if (flx_interface_match(i, e->interface, e->protocol) && flx_entry_established(s, e, i))
- flx_interface_post_response(i, a, e->record, e->flags & FLX_SERVER_ENTRY_UNIQUE, FALSE);
+ for (e = s->entries; e; e = e->entries_next)
+ if (!e->dead && flx_key_pattern_match(k, e->record->key) && flx_entry_registered(s, e, i))
+ flx_interface_post_response(i, a, e->record, e->flags & FLX_ENTRY_UNIQUE, FALSE);
} else {
/* Handle all other queries */
- for (e = g_hash_table_lookup(s->rrset_by_key, k); e; e = e->by_key_next)
- if (flx_interface_match(i, e->interface, e->protocol) && flx_entry_established(s, e, i))
- flx_interface_post_response(i, a, e->record, e->flags & FLX_SERVER_ENTRY_UNIQUE, FALSE);
+ for (e = g_hash_table_lookup(s->entries_by_key, k); e; e = e->by_key_next)
+ if (!e->dead && flx_entry_registered(s, e, i))
+ flx_interface_post_response(i, a, e->record, e->flags & FLX_ENTRY_UNIQUE, FALSE);
}
}
+static void withdraw_entry(flxServer *s, flxEntry *e) {
+ g_assert(s);
+ g_assert(e);
+
+ e->dead = TRUE;
+ s->need_entry_cleanup = TRUE;
+
+ flx_goodbye_entry(s, e, FALSE);
+
+ if (e->group)
+ flx_entry_group_run_callback(e->group, FLX_ENTRY_GROUP_COLLISION);
+}
+
+static void incoming_probe(flxServer *s, flxRecord *record, flxInterface *i) {
+ flxEntry *e, *n;
+ gchar *t;
+
+ g_assert(s);
+ g_assert(record);
+ g_assert(i);
+
+ t = flx_record_to_string(record);
+
+ for (e = g_hash_table_lookup(s->entries_by_key, record->key); e; e = n) {
+ n = e->by_key_next;
+
+ if (e->dead || !flx_record_equal_no_ttl(record, e->record))
+ continue;
+
+ if (flx_entry_registering(s, e, i)) {
+
+ if (flx_record_lexicographical_compare(record, e->record) > 0) {
+ withdraw_entry(s, e);
+ g_message("Recieved conflicting probe [%s]. Local host lost. Withdrawing.", t);
+ } else
+ g_message("Recieved conflicting probe [%s]. Local host won.", t);
+ }
+ }
+
+ g_free(t);
+}
+
static void handle_query(flxServer *s, flxDnsPacket *p, flxInterface *i, const flxAddress *a) {
guint n;
g_assert(i);
g_assert(a);
+ /* Handle the questions */
for (n = flx_dns_packet_get_field(p, FLX_DNS_FIELD_QDCOUNT); n > 0; n --) {
flxKey *key;
if (!(key = flx_dns_packet_consume_key(p))) {
- g_warning("Packet too short");
+ g_warning("Packet too short (1)");
return;
}
flx_packet_scheduler_incoming_known_answer(i->scheduler, record, a);
flx_record_unref(record);
}
+
+ /* Probe record */
+ for (n = flx_dns_packet_get_field(p, FLX_DNS_FIELD_NSCOUNT); n > 0; n --) {
+ flxRecord *record;
+ gboolean unique = FALSE;
+
+ if (!(record = flx_dns_packet_consume_record(p, &unique))) {
+ g_warning("Packet too short (3)");
+ return;
+ }
+
+ if (record->key->type != FLX_DNS_TYPE_ANY)
+ incoming_probe(s, record, i);
+
+ flx_record_unref(record);
+ }
+}
+
+static gboolean handle_conflict(flxServer *s, flxInterface *i, flxRecord *record, const flxAddress *a) {
+ gboolean valid = TRUE;
+ flxEntry *e, *n;
+ gchar *t;
+
+ g_assert(s);
+ g_assert(i);
+ g_assert(record);
+
+ t = flx_record_to_string(record);
+
+ for (e = g_hash_table_lookup(s->entries_by_key, record->key); e; e = n) {
+ n = e->by_key_next;
+
+ if (e->dead)
+ continue;
+
+ if (flx_entry_registered(s, e, i)) {
+
+ gboolean equal = flx_record_equal_no_ttl(record, e->record);
+
+ /* Check whether there is a unique record conflict */
+ if (!equal && (e->flags & FLX_ENTRY_UNIQUE)) {
+
+ /* The lexicographically later data wins. */
+ if (flx_record_lexicographical_compare(record, e->record) > 0) {
+ withdraw_entry(s, e);
+ g_message("Recieved conflicting record [%s]. Local host lost. Withdrawing.", t);
+ } else {
+ /* Tell the other host that our entry is lexicographically later */
+ valid = FALSE;
+ flx_interface_post_response(i, a, e->record, e->flags & FLX_ENTRY_UNIQUE, TRUE);
+ g_message("Recieved conflicting record [%s]. Local host won. Refreshing.", t);
+ }
+
+ /* Check wheter there is a TTL conflict */
+ } else if (equal && record->ttl <= e->record->ttl/2) {
+ /* Correct the TTL */
+ valid = FALSE;
+ flx_interface_post_response(i, a, e->record, e->flags & FLX_ENTRY_UNIQUE, TRUE);
+ g_message("Recieved record with bad TTL [%s]. Refreshing.", t);
+ }
+ }
+ }
+
+ g_free(t);
+
+ return valid;
}
static void handle_response(flxServer *s, flxDnsPacket *p, flxInterface *i, const flxAddress *a) {
gchar *txt;
if (!(record = flx_dns_packet_consume_record(p, &cache_flush))) {
- g_warning("Packet too short (3)");
+ g_warning("Packet too short (4)");
return;
}
- if (record->key->type == FLX_DNS_TYPE_ANY)
+ if (record->key->type != FLX_DNS_TYPE_ANY) {
continue;
- g_message("Handling response: %s", txt = flx_record_to_string(record));
- g_free(txt);
-
- flx_cache_update(i->cache, record, cache_flush, a);
-
- flx_packet_scheduler_incoming_response(i->scheduler, record);
+ g_message("Handling response: %s", txt = flx_record_to_string(record));
+ g_free(txt);
+
+ if (handle_conflict(s, i, record, a)) {
+ flx_cache_update(i->cache, record, cache_flush, a);
+ flx_packet_scheduler_incoming_response(i->scheduler, record);
+ }
+ }
+
flx_record_unref(record);
}
}
if (ttl != 255) {
g_warning("Recieved packet with invalid TTL on interface '%s.%i'.", i->hardware->name, i->protocol);
- return;
+ if (!s->ignore_bad_ttl)
+ return;
}
if (sa->sa_family == AF_INET6) {
}
}
-static gboolean work(flxServer *s) {
+static void work(flxServer *s) {
struct sockaddr_in6 sa6;
struct sockaddr_in sa;
flxDnsPacket *p;
flx_dns_packet_free(p);
}
}
-
- return TRUE;
}
static gboolean prepare_func(GSource *source, gint *timeout) {
s = *((flxServer**) (((guint8*) source) + sizeof(GSource)));
g_assert(s);
-
- return work(s);
+
+ work(s);
+ cleanup_dead(s);
+
+ return TRUE;
}
static void add_default_entries(flxServer *s) {
uname(&utsname);
r->data.hinfo.cpu = g_strdup(g_strup(utsname.machine));
r->data.hinfo.os = g_strdup(g_strup(utsname.sysname));
- flx_server_add(s, 0, 0, AF_UNSPEC, FLX_SERVER_ENTRY_UNIQUE, r);
+ flx_server_add(s, NULL, 0, AF_UNSPEC, FLX_ENTRY_UNIQUE | FLX_ENTRY_NOANNOUNCE | FLX_ENTRY_NOPROBE, r);
flx_record_unref(r);
/* Add localhost entries */
flx_address_parse("127.0.0.1", AF_INET, &a);
- flx_server_add_address(s, 0, 0, AF_UNSPEC, FLX_SERVER_ENTRY_UNIQUE|FLX_SERVER_ENTRY_NOPROBE|FLX_SERVER_ENTRY_NOANNOUNCE, "localhost", &a);
+ flx_server_add_address(s, NULL, 0, AF_UNSPEC, FLX_ENTRY_UNIQUE|FLX_ENTRY_NOPROBE|FLX_ENTRY_NOANNOUNCE, "localhost", &a);
flx_address_parse("::1", AF_INET6, &a);
- flx_server_add_address(s, 0, 0, AF_UNSPEC, FLX_SERVER_ENTRY_UNIQUE|FLX_SERVER_ENTRY_NOPROBE|FLX_SERVER_ENTRY_NOANNOUNCE, "ip6-localhost", &a);
+ flx_server_add_address(s, NULL, 0, AF_UNSPEC, FLX_ENTRY_UNIQUE|FLX_ENTRY_NOPROBE|FLX_ENTRY_NOANNOUNCE, "ip6-localhost", &a);
}
flxServer *flx_server_new(GMainContext *c) {
s = g_new(flxServer, 1);
+ s->ignore_bad_ttl = FALSE;
+ s->need_entry_cleanup = s->need_group_cleanup = FALSE;
+
s->fd_ipv4 = flx_open_socket_ipv4();
- s->fd_ipv6 = flx_open_socket_ipv6();
+ s->fd_ipv6 = -1 /*flx_open_socket_ipv6() */;
if (s->fd_ipv6 < 0 && s->fd_ipv4 < 0) {
- g_critical("Failed to create sockets.\n");
+ g_critical("Failed to create IP sockets.\n");
g_free(s);
return NULL;
}
else
s->context = g_main_context_default();
- s->current_id = 1;
-
- FLX_LLIST_HEAD_INIT(flxServerEntry, s->entries);
- s->rrset_by_id = g_hash_table_new(g_int_hash, g_int_equal);
- s->rrset_by_key = g_hash_table_new((GHashFunc) flx_key_hash, (GEqualFunc) flx_key_equal);
+ FLX_LLIST_HEAD_INIT(flxEntry, s->entries);
+ s->entries_by_key = g_hash_table_new((GHashFunc) flx_key_hash, (GEqualFunc) flx_key_equal);
+ FLX_LLIST_HEAD_INIT(flxGroup, s->groups);
FLX_LLIST_HEAD_INIT(flxSubscription, s->subscriptions);
s->subscription_hashtable = g_hash_table_new((GHashFunc) flx_key_hash, (GEqualFunc) flx_key_equal);
- s->monitor = flx_interface_monitor_new(s);
- s->time_event_queue = flx_time_event_queue_new(s->context);
-
/* Get host name */
hn = flx_get_host_name();
hn[strcspn(hn, ".")] = 0;
s->hostname = g_strdup_printf("%s.local.", hn);
g_free(hn);
+ s->time_event_queue = flx_time_event_queue_new(s->context, G_PRIORITY_DEFAULT+10); /* Slightly less priority than the FDs */
+ s->monitor = flx_interface_monitor_new(s);
+ flx_interface_monitor_sync(s->monitor);
add_default_entries(s);
-
+
+ /* Prepare IO source registration */
s->source = g_source_new(&source_funcs, sizeof(GSource) + sizeof(flxServer*));
*((flxServer**) (((guint8*) s->source) + sizeof(GSource))) = s;
g_source_add_poll(s->source, &s->pollfd_ipv6);
g_source_attach(s->source, s->context);
-
+
return s;
}
void flx_server_free(flxServer* s) {
g_assert(s);
+ while(s->entries)
+ free_entry(s, s->entries);
+
flx_interface_monitor_free(s->monitor);
-
- flx_server_remove(s, 0);
+
+ while (s->groups)
+ free_group(s, s->groups);
while (s->subscriptions)
flx_subscription_free(s->subscriptions);
g_hash_table_destroy(s->subscription_hashtable);
-
- g_hash_table_destroy(s->rrset_by_id);
- g_hash_table_destroy(s->rrset_by_key);
+
+ g_hash_table_destroy(s->entries_by_key);
flx_time_event_queue_free(s->time_event_queue);
g_free(s);
}
-gint flx_server_get_next_id(flxServer *s) {
- g_assert(s);
-
- return s->current_id++;
-}
-
void flx_server_add(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
- flxServerEntryFlags flags,
+ flxEntryFlags flags,
flxRecord *r) {
- flxServerEntry *e, *t;
+ flxEntry *e, *t;
g_assert(s);
g_assert(r);
g_assert(r->key->type != FLX_DNS_TYPE_ANY);
- e = g_new(flxServerEntry, 1);
+ e = g_new(flxEntry, 1);
+ e->server = s;
e->record = flx_record_ref(r);
- e->id = id;
+ e->group = g;
e->interface = interface;
e->protocol = protocol;
e->flags = flags;
+ e->dead = FALSE;
FLX_LLIST_HEAD_INIT(flxAnnouncement, e->announcements);
- FLX_LLIST_PREPEND(flxServerEntry, entry, s->entries, e);
+ FLX_LLIST_PREPEND(flxEntry, entries, s->entries, e);
- /* Insert into hash table indexed by id */
- t = g_hash_table_lookup(s->rrset_by_id, &e->id);
- FLX_LLIST_PREPEND(flxServerEntry, by_id, t, e);
- g_hash_table_replace(s->rrset_by_id, &e->id, t);
-
/* Insert into hash table indexed by name */
- t = g_hash_table_lookup(s->rrset_by_key, e->record->key);
- FLX_LLIST_PREPEND(flxServerEntry, by_key, t, e);
- g_hash_table_replace(s->rrset_by_key, e->record->key, t);
+ t = g_hash_table_lookup(s->entries_by_key, e->record->key);
+ FLX_LLIST_PREPEND(flxEntry, by_key, t, e);
+ g_hash_table_replace(s->entries_by_key, e->record->key, t);
+
+ /* Insert into group list */
+ if (g)
+ FLX_LLIST_PREPEND(flxEntry, by_group, g->entries, e);
flx_announce_entry(s, e);
}
-const flxRecord *flx_server_iterate(flxServer *s, gint id, void **state) {
- flxServerEntry **e = (flxServerEntry**) state;
+const flxRecord *flx_server_iterate(flxServer *s, flxEntryGroup *g, void **state) {
+ flxEntry **e = (flxEntry**) state;
g_assert(s);
g_assert(e);
- if (e)
- *e = id > 0 ? (*e)->by_id_next : (*e)->entry_next;
- else
- *e = id > 0 ? g_hash_table_lookup(s->rrset_by_id, &id) : s->entries;
+ if (!*e)
+ *e = g ? g->entries : s->entries;
+
+ while (*e && (*e)->dead)
+ *e = g ? (*e)->by_group_next : (*e)->entries_next;
if (!*e)
return NULL;
return flx_record_ref((*e)->record);
}
-static void free_entry(flxServer*s, flxServerEntry *e) {
- flxServerEntry *t;
-
- g_assert(e);
-
- flx_goodbye_entry(s, e, TRUE);
-
- /* Remove from linked list */
- FLX_LLIST_REMOVE(flxServerEntry, entry, s->entries, e);
-
- /* Remove from hash table indexed by id */
- t = g_hash_table_lookup(s->rrset_by_id, &e->id);
- FLX_LLIST_REMOVE(flxServerEntry, by_id, t, e);
- if (t)
- g_hash_table_replace(s->rrset_by_id, &t->id, t);
- else
- g_hash_table_remove(s->rrset_by_id, &e->id);
-
- /* Remove from hash table indexed by name */
- t = g_hash_table_lookup(s->rrset_by_key, e->record->key);
- FLX_LLIST_REMOVE(flxServerEntry, by_key, t, e);
- if (t)
- g_hash_table_replace(s->rrset_by_key, t->record->key, t);
- else
- g_hash_table_remove(s->rrset_by_key, e->record->key);
-
- flx_record_unref(e->record);
- g_free(e);
-}
-
-void flx_server_remove(flxServer *s, gint id) {
- g_assert(s);
-
- if (id <= 0) {
- while (s->entries)
- free_entry(s, s->entries);
- } else {
- flxServerEntry *e;
-
- while ((e = g_hash_table_lookup(s->rrset_by_id, &id)))
- free_entry(s, e);
- }
-}
-
void flx_server_dump(flxServer *s, FILE *f) {
- flxServerEntry *e;
+ flxEntry *e;
g_assert(s);
g_assert(f);
fprintf(f, "\n;;; ZONE DUMP FOLLOWS ;;;\n");
- for (e = s->entries; e; e = e->entry_next) {
+ for (e = s->entries; e; e = e->entries_next) {
gchar *t;
+ if (e->dead)
+ continue;
+
t = flx_record_to_string(e->record);
- fprintf(f, "%s\n", t);
+ fprintf(f, "%s ; iface=%i proto=%i\n", t, e->interface, e->protocol);
g_free(t);
}
void flx_server_add_ptr(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
- flxServerEntryFlags flags,
+ flxEntryFlags flags,
const gchar *name,
const gchar *dest) {
r = flx_record_new_full(name ? name : s->hostname, FLX_DNS_CLASS_IN, FLX_DNS_TYPE_PTR);
r->data.ptr.name = flx_normalize_name(dest);
- flx_server_add(s, id, interface, protocol, flags, r);
+ flx_server_add(s, g, interface, protocol, flags, r);
flx_record_unref(r);
-
}
void flx_server_add_address(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
- flxServerEntryFlags flags,
+ flxEntryFlags flags,
const gchar *name,
flxAddress *a) {
r = flx_record_new_full(name, FLX_DNS_CLASS_IN, FLX_DNS_TYPE_A);
r->data.a.address = a->data.ipv4;
- flx_server_add(s, id, interface, protocol, flags, r);
+ flx_server_add(s, g, interface, protocol, flags, r);
flx_record_unref(r);
reverse = flx_reverse_lookup_name_ipv4(&a->data.ipv4);
g_assert(reverse);
- flx_server_add_ptr(s, id, interface, protocol, flags, reverse, name);
+ flx_server_add_ptr(s, g, interface, protocol, flags, reverse, name);
g_free(reverse);
} else {
r = flx_record_new_full(name, FLX_DNS_CLASS_IN, FLX_DNS_TYPE_AAAA);
r->data.aaaa.address = a->data.ipv6;
- flx_server_add(s, id, interface, protocol, flags, r);
+ flx_server_add(s, g, interface, protocol, flags, r);
flx_record_unref(r);
reverse = flx_reverse_lookup_name_ipv6_arpa(&a->data.ipv6);
g_assert(reverse);
- flx_server_add_ptr(s, id, interface, protocol, flags, reverse, name);
+ flx_server_add_ptr(s, g, interface, protocol, flags, reverse, name);
g_free(reverse);
reverse = flx_reverse_lookup_name_ipv6_int(&a->data.ipv6);
g_assert(reverse);
- flx_server_add_ptr(s, id, interface, protocol, flags, reverse, name);
+ flx_server_add_ptr(s, g, interface, protocol, flags, reverse, name);
g_free(reverse);
}
void flx_server_add_text_va(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
- flxServerEntryFlags flags,
+ flxEntryFlags flags,
const gchar *name,
va_list va) {
r = flx_record_new_full(name ? name : s->hostname, FLX_DNS_CLASS_IN, FLX_DNS_TYPE_TXT);
r->data.txt.string_list = flx_string_list_new_va(va);
- flx_server_add(s, id, interface, protocol, flags, r);
+ flx_server_add(s, g, interface, protocol, flags, r);
flx_record_unref(r);
}
void flx_server_add_text(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
- flxServerEntryFlags flags,
+ flxEntryFlags flags,
const gchar *name,
...) {
g_assert(s);
va_start(va, name);
- flx_server_add_text_va(s, id, interface, protocol, flags, name, va);
+ flx_server_add_text_va(s, g, interface, protocol, flags, name, va);
va_end(va);
}
void flx_server_add_service_va(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
const gchar *type,
snprintf(ptr_name, sizeof(ptr_name), "%s.%s", type, domain);
snprintf(svc_name, sizeof(svc_name), "%s.%s.%s", ename, type, domain);
- flx_server_add_ptr(s, id, interface, protocol, FALSE, ptr_name, svc_name);
+ flx_server_add_ptr(s, g, interface, protocol, FALSE, ptr_name, svc_name);
r = flx_record_new_full(svc_name, FLX_DNS_CLASS_IN, FLX_DNS_TYPE_SRV);
r->data.srv.priority = 0;
r->data.srv.weight = 0;
r->data.srv.port = port;
r->data.srv.name = flx_normalize_name(host);
- flx_server_add(s, id, interface, protocol, TRUE, r);
+ flx_server_add(s, g, interface, protocol, TRUE, r);
flx_record_unref(r);
- flx_server_add_text_va(s, id, interface, protocol, FALSE, svc_name, va);
+ flx_server_add_text_va(s, g, interface, protocol, FALSE, svc_name, va);
snprintf(enum_ptr, sizeof(enum_ptr), "_services._dns-sd._udp.%s", domain);
- flx_server_add_ptr(s, id, interface, protocol, FALSE, enum_ptr, ptr_name);
+ flx_server_add_ptr(s, g, interface, protocol, FALSE, enum_ptr, ptr_name);
}
void flx_server_add_service(
flxServer *s,
- gint id,
+ flxEntryGroup *g,
gint interface,
guchar protocol,
const gchar *type,
g_assert(name);
va_start(va, port);
- flx_server_add_service_va(s, id, interface, protocol, type, name, domain, host, port, va);
+ flx_server_add_service_va(s, g, interface, protocol, type, name, domain, host, port, va);
va_end(va);
}
flx_interface_monitor_walk(s->monitor, interface, protocol, post_response_callback, &tmpdata);
}
+
+void flx_entry_group_run_callback(flxEntryGroup *g, flxEntryGroupStatus status) {
+ g_assert(g);
+
+ if (g->callback) {
+ g->callback(g->server, g, status, g->userdata);
+ return;
+ }
+
+ if (status == FLX_ENTRY_GROUP_COLLISION)
+ flx_entry_group_free(g);
+
+ /* Ignore the rest */
+}
+
+flxEntryGroup *flx_entry_group_new(flxServer *s, flxEntryGroupCallback callback, gpointer userdata) {
+ flxEntryGroup *g;
+
+ g_assert(s);
+
+ g = g_new(flxEntryGroup, 1);
+ g->server = s;
+ g->callback = callback;
+ g->userdata = userdata;
+ g->dead = FALSE;
+ g->status = FLX_ENTRY_GROUP_UNCOMMITED;
+ g->n_probing = 0;
+ FLX_LLIST_HEAD_INIT(flxEntry, g->entries);
+
+ FLX_LLIST_PREPEND(flxEntryGroup, groups, s->groups, g);
+ return g;
+}
+
+void flx_entry_group_free(flxEntryGroup *g) {
+ g_assert(g);
+ g_assert(g->server);
+
+ g->dead = TRUE;
+ g->server->need_group_cleanup = TRUE;
+}
+
+void flx_entry_group_commit(flxEntryGroup *g) {
+ flxEntry *e;
+
+ g_assert(g);
+ g_assert(!g->dead);
+
+ if (g->status != FLX_ENTRY_GROUP_UNCOMMITED)
+ return;
+
+ flx_entry_group_run_callback(g, g->status = FLX_ENTRY_GROUP_REGISTERING);
+ flx_announce_group(g->server, g);
+ flx_entry_group_check_probed(g, FALSE);
+}
+
+gboolean flx_entry_commited(flxEntry *e) {
+ g_assert(e);
+ g_assert(!e->dead);
+
+ return !e->group ||
+ e->group->status == FLX_ENTRY_GROUP_REGISTERING ||
+ e->group->status == FLX_ENTRY_GROUP_ESTABLISHED;
+}
+
+flxEntryGroupStatus flx_entry_group_get_status(flxEntryGroup *g) {
+ g_assert(g);
+ g_assert(!g->dead);
+
+ return g->status;
+}
#ifndef fooflxserverhfoo
#define fooflxserverhfoo
-typedef struct _flxServerEntry flxServerEntry;
-
#include "flx.h"
#include "iface.h"
#include "prioq.h"
#include "announce.h"
#include "subscribe.h"
-struct _flxServerEntry {
+struct _flxEntry {
+ flxServer *server;
+ flxEntryGroup *group;
+
+ gboolean dead;
+
+ flxEntryFlags flags;
flxRecord *record;
- gint id;
gint interface;
guchar protocol;
- flxServerEntryFlags flags;
-
- FLX_LLIST_FIELDS(flxServerEntry, entry);
- FLX_LLIST_FIELDS(flxServerEntry, by_key);
- FLX_LLIST_FIELDS(flxServerEntry, by_id);
+ FLX_LLIST_FIELDS(flxEntry, entries);
+ FLX_LLIST_FIELDS(flxEntry, by_key);
+ FLX_LLIST_FIELDS(flxEntry, by_group);
FLX_LLIST_HEAD(flxAnnouncement, announcements);
};
+struct _flxEntryGroup {
+ flxServer *server;
+ gboolean dead;
+
+ flxEntryGroupStatus status;
+ gpointer userdata;
+ flxEntryGroupCallback callback;
+
+ guint n_probing;
+
+ FLX_LLIST_FIELDS(flxEntryGroup, groups);
+ FLX_LLIST_HEAD(flxEntry, entries);
+};
+
struct _flxServer {
GMainContext *context;
flxInterfaceMonitor *monitor;
- gint current_id;
-
- FLX_LLIST_HEAD(flxServerEntry, entries);
- GHashTable *rrset_by_id;
- GHashTable *rrset_by_key;
+ FLX_LLIST_HEAD(flxEntry, entries);
+ GHashTable *entries_by_key;
+ FLX_LLIST_HEAD(flxEntryGroup, groups);
+
FLX_LLIST_HEAD(flxSubscription, subscriptions);
GHashTable *subscription_hashtable;
+ gboolean need_entry_cleanup, need_group_cleanup;
+
flxTimeEventQueue *time_event_queue;
gchar *hostname;
GPollFD pollfd_ipv4, pollfd_ipv6;
GSource *source;
-
+
+ gboolean ignore_bad_ttl;
};
-gboolean flx_server_entry_match_interface(flxServerEntry *e, flxInterface *i);
+gboolean flx_server_entry_match_interface(flxEntry *e, flxInterface *i);
+
+void flx_server_post_query(flxServer *s, gint interface, guchar protocol, flxKey *key);
+void flx_server_post_response(flxServer *s, gint interface, guchar protocol, flxRecord *record, gboolean flush_cache);
+
+void flx_entry_group_run_callback(flxEntryGroup *g, flxEntryGroupStatus state);
+
+gboolean flx_entry_commited(flxEntry *e);
#endif
#ifndef foosubscribehfoo
#define foosubscribehfoo
-typedef struct _flxSubscription flxSubscription;
-
#include "llist.h"
+#include "flx.h"
+#include "subscribe.h"
+#include "timeeventq.h"
#include "server.h"
-typedef enum {
- FLX_SUBSCRIPTION_NEW,
- FLX_SUBSCRIPTION_REMOVE,
- FLX_SUBSCRIPTION_CHANGE
-} flxSubscriptionEvent;
-
-typedef void (*flxSubscriptionCallback)(flxSubscription *s, flxRecord *record, gint interface, guchar protocol, flxSubscriptionEvent event, gpointer userdata);
-
struct _flxSubscription {
flxServer *server;
flxKey *key;
FLX_LLIST_FIELDS(flxSubscription, by_key);
};
-flxSubscription *flx_subscription_new(flxServer *s, flxKey *key, gint interface, guchar protocol, flxSubscriptionCallback callback, gpointer userdata);
-void flx_subscription_free(flxSubscription *s);
-
void flx_subscription_notify(flxServer *s, flxInterface *i, flxRecord *record, flxSubscriptionEvent event);
gboolean flx_is_subscribed(flxServer *s, flxKey *k);
return TRUE;
}
-flxTimeEventQueue* flx_time_event_queue_new(GMainContext *context) {
+flxTimeEventQueue* flx_time_event_queue_new(GMainContext *context, gint priority) {
flxTimeEventQueue *q;
static GSourceFuncs source_funcs = {
q = (flxTimeEventQueue*) g_source_new(&source_funcs, sizeof(flxTimeEventQueue));
q->prioq = flx_prio_queue_new(compare);
+ g_source_set_priority((GSource*) q, priority);
+
g_source_attach(&q->source, context);
return q;
flxPrioQueue *prioq;
};
-flxTimeEventQueue* flx_time_event_queue_new(GMainContext *context);
+flxTimeEventQueue* flx_time_event_queue_new(GMainContext *context, gint priority);
void flx_time_event_queue_free(flxTimeEventQueue *q);
flxTimeEvent* flx_time_event_queue_add(flxTimeEventQueue *q, const GTimeVal *timeval, void (*callback)(flxTimeEvent *e, void *userdata), void *userdata);
todo:
-* Probing/Conflict resolution
-* uniqueness
-
-* defend our entries on incoming goodbye
-
* Unicast responses/queries
* Legacy unicast
-* remove expression "rrset" from source files
-
* add SRV and TXT records referenced from PTR records automatically to packet
* add A and AAAA records referenced from SRV records automatically to packet
* allow NULL bytes in TXT records
done:
+* Probing/Conflict resolution
+* uniqueness
* respect escaping in name serialization
* really send goodbye packets
* refresh subscribed records only
* Known-Answer suppression server part
* make flx_server_add_text() and flx_server_add_service() variadic functions
* name compression
+* remove expression "rrset" from source files
+* defend our entries on incoming goodbye
return flx_timeval_diff(&now, a);
}
-gboolean flx_domain_equal(const gchar *a, const gchar *b) {
+gboolean flx_domain_cmp(const gchar *a, const gchar *b) {
int escaped_a = 0, escaped_b = 0;
g_assert(a);
g_assert(b);
/* Check for string end */
if (*a == 0 && *b == 0)
- return TRUE;
+ return 0;
if (*a == 0 && !escaped_b && *b == '.' && *(b+1) == 0)
- return TRUE;
+ return 0;
if (!escaped_a && *a == '.' && *(a+1) == 0 && *b == 0)
- return TRUE;
+ return 0;
/* Compare characters */
if (escaped_a == escaped_b && *a != *b)
- return FALSE;
-
+ return *a < *b ? -1 : 1;
/* Next characters */
a++;
}
}
+gboolean flx_domain_equal(const gchar *a, const gchar *b) {
+ return flx_domain_cmp(a, b) == 0;
+}
+
guint flx_domain_hash(const gchar *p) {
char t[256];
strncpy(t, p, sizeof(t)-1);
gint flx_age(const GTimeVal *a);
guint flx_domain_hash(const gchar *p);
+gboolean flx_domain_cmp(const gchar *a, const gchar *b);
gboolean flx_domain_equal(const gchar *a, const gchar *b);
void flx_hexdump(gconstpointer p, guint size);
projects / catta / commitdiff