]> git.ipfire.org Git - thirdparty/unbound.git/commitdiff
mesh work, now coded mesh service.
authorWouter Wijngaards <wouter@nlnetlabs.nl>
Mon, 25 Jun 2007 13:54:03 +0000 (13:54 +0000)
committerWouter Wijngaards <wouter@nlnetlabs.nl>
Mon, 25 Jun 2007 13:54:03 +0000 (13:54 +0000)
git-svn-id: file:///svn/unbound/trunk@422 be551aaa-1e26-0410-a405-d3ace91eadb9

daemon/worker.c
doc/Changelog
services/mesh.c
services/mesh.h
util/data/msgencode.c
util/data/msgencode.h
util/module.h

index c4683f72606e36a731419146866287544e8dc4ef..4c102350db46a967592b223d9b01ae97d0fc25f5 100644 (file)
@@ -152,43 +152,12 @@ req_release(struct work_query* w)
                (int)worker->num_requests);
 }
 
-/** create error and fill into buffer */
-static void
-replyerror_fillbuf(int r, struct comm_reply* repinfo, uint16_t id,
-       uint16_t qflags, struct query_info* qinfo)
-{
-       ldns_buffer* buf = repinfo->c->buffer;
-       uint16_t flags;
-       verbose(VERB_DETAIL, "reply with error");
-       
-       ldns_buffer_clear(buf);
-       ldns_buffer_write(buf, &id, sizeof(uint16_t));
-       flags = (uint16_t)(BIT_QR | BIT_RA | r); /* QR and retcode*/
-       flags |= (qflags & (BIT_RD|BIT_CD)); /* copy RD and CD bit */
-       ldns_buffer_write_u16(buf, flags);
-       flags = 1;
-       ldns_buffer_write_u16(buf, flags);
-       flags = 0;
-       ldns_buffer_write(buf, &flags, sizeof(uint16_t));
-       ldns_buffer_write(buf, &flags, sizeof(uint16_t));
-       ldns_buffer_write(buf, &flags, sizeof(uint16_t));
-       ldns_buffer_write(buf, qinfo->qname, qinfo->qname_len);
-       ldns_buffer_write_u16(buf, qinfo->qtype);
-       ldns_buffer_write_u16(buf, qinfo->qclass);
-       ldns_buffer_flip(buf);
-}
-
 /** reply to query with given error code */
 static void 
 replyerror(int r, struct work_query* w)
 {
-       w->state.edns.edns_version = EDNS_ADVERTISED_VERSION;
-       w->state.edns.udp_size = EDNS_ADVERTISED_SIZE;
-       w->state.edns.ext_rcode = 0;
-       w->state.edns.bits &= EDNS_DO;
-       replyerror_fillbuf(r, &w->query_reply, w->query_id, 
-               w->state.query_flags, &w->state.qinfo);
-       attach_edns_record(w->query_reply.c->buffer, &w->state.edns);
+       error_encode(w->query_reply.c->buffer, r, &w->state.qinfo, 
+               w->query_id, w->state.query_flags, &w->state.edns);
        comm_point_send_reply(&w->query_reply);
        req_release(w);
        query_info_clear(&w->state.qinfo);
@@ -512,8 +481,8 @@ answer_from_cache(struct worker* worker, struct lruhash_entry* e, uint16_t id,
        if(!reply_info_answer_encode(&mrentry->key, rep, id, flags, 
                repinfo->c->buffer, timenow, 1, worker->scratchpad,
                udpsize, edns, (int)(edns->bits & EDNS_DO) )) {
-               replyerror_fillbuf(LDNS_RCODE_SERVFAIL, repinfo, id,
-                       flags, &mrentry->key);
+               error_encode(repinfo->c->buffer, LDNS_RCODE_SERVFAIL, 
+                       &mrentry->key, id, flags, edns);
        }
        /* cannot send the reply right now, because blocking network syscall
         * is bad while holding locks. */
@@ -581,9 +550,9 @@ worker_handle_request(struct comm_point* c, void* arg, int error,
                edns.udp_size = EDNS_ADVERTISED_SIZE;
                edns.bits &= EDNS_DO;
                verbose(VERB_ALGO, "query with bad edns version.");
-               replyerror_fillbuf(EDNS_RCODE_BADVERS&0xf, repinfo, 
+               error_encode(c->buffer, EDNS_RCODE_BADVERS&0xf, &qinfo,
                        *(uint16_t*)ldns_buffer_begin(c->buffer),
-                       ldns_buffer_read_u16_at(c->buffer, 2), &qinfo);
+                       ldns_buffer_read_u16_at(c->buffer, 2), NULL);
                attach_edns_record(c->buffer, &edns);
                return 1;
        }
@@ -871,7 +840,11 @@ worker_init(struct worker* worker, struct config_file *cfg,
        worker->env.rnd = worker->rndstate;
        worker->env.scratch = worker->scratchpad;
        worker->env.mesh = mesh_create(worker->daemon->num_modules,
-               worker->daemon->modfunc);
+               worker->daemon->modfunc, &worker->env);
+       worker->env.detach_subs = &mesh_detach_subs;
+       worker->env.attach_sub = &mesh_attach_sub;
+       worker->env.query_done = &mesh_query_done;
+       worker->env.walk_supers = &mesh_walk_supers;
        if(!worker->env.mesh) {
                worker_delete(worker);
                return 0;
index 601be3db1c0494f433e43b15744e4ed66b4034e8..34ff75a6b16f04134c054b2fdf9579a544f456c1 100644 (file)
@@ -1,3 +1,7 @@
+25 June 2007: Wouter
+       - more mesh work.
+       - error encode routine for ease.
+
 22 June 2007: Wouter
        - removed unused _node iterator value from rbtree_t. Takes up space.
        - iterator can handle querytargets state without a delegation point
index ef185f16236129f2628a02b896bb705b36dd338f..a266898a4b757e18c61995089f80a5b6b60c0276 100644 (file)
  */
 #include "config.h"
 #include "services/mesh.h"
+#include "services/outbound_list.h"
 #include "util/log.h"
 #include "util/net_help.h"
 #include "util/module.h"
 #include "util/region-allocator.h"
+#include "util/data/msgencode.h"
 
 /** compare two mesh_states */
 static int
@@ -79,7 +81,8 @@ mesh_state_ref_compare(const void* ap, const void* bp)
 }
 
 struct mesh_area* 
-mesh_create(int num_modules, struct module_func_block** modfunc)
+mesh_create(int num_modules, struct module_func_block** modfunc,
+       struct module_env* env)
 {
        struct mesh_area* mesh = calloc(1, sizeof(struct mesh_area));
        if(!mesh) {
@@ -88,6 +91,7 @@ mesh_create(int num_modules, struct module_func_block** modfunc)
        }
        mesh->num_modules = num_modules;
        mesh->modfunc = modfunc;
+       mesh->env = env;
        rbtree_init(&mesh->run, &mesh_state_compare);
        rbtree_init(&mesh->all, &mesh_state_compare);
        mesh->num_reply_addrs = 0;
@@ -109,6 +113,65 @@ mesh_delete(struct mesh_area* mesh)
        free(mesh);
 }
 
+void mesh_new_client(struct mesh_area* mesh, struct query_info* qinfo,
+        uint16_t qflags, struct edns_data* edns, struct comm_reply* rep,
+        uint16_t qid)
+{
+       struct mesh_state* s = mesh_area_find(mesh, qinfo, qflags, 0);
+       int was_detached = 0;
+       int was_noreply = 0;
+       int added = 0;
+       /* see if it already exists, if not, create one */
+       if(!s) {
+               struct rbnode_t* n;
+               s = mesh_state_create(mesh->env,qinfo, qflags, 0);
+               if(!s) {
+                       log_err("mesh_state_create: out of memory; SERVFAIL");
+                       error_encode(rep->c->buffer, LDNS_RCODE_SERVFAIL,
+                               qinfo, qid, qflags, edns);
+                       comm_point_send_reply(rep);
+                       return;
+               }
+               n = rbtree_insert(&mesh->all, &s->node);
+               log_assert(n != NULL);
+               /* set detached (it is now) */
+               mesh->num_detached_states++;
+               added = 1;
+       }
+       if(!s->reply_list && s->super_set.count == 0)
+               was_detached = 1;
+       if(!s->reply_list)
+               was_noreply = 1;
+       /* add reply to s */
+       if(!mesh_state_add_reply(s, edns, rep, qid, qflags)) {
+                       log_err("mesh_new_client: out of memory; SERVFAIL");
+                       error_encode(rep->c->buffer, LDNS_RCODE_SERVFAIL,
+                               qinfo, qid, qflags, edns);
+                       comm_point_send_reply(rep);
+                       if(added)
+                               mesh_state_delete(s);
+                       return;
+       }
+       /* update statistics */
+       if(was_detached) {
+               log_assert(mesh->num_detached_states > 0);
+               mesh->num_detached_states--;
+       }
+       if(was_noreply) {
+               mesh->num_reply_states ++;
+       }
+       if(added)
+               mesh_run(mesh, s, module_event_new, NULL);
+}
+
+void mesh_report_reply(struct mesh_area* mesh, struct outbound_entry* e,
+        int is_ok, struct comm_reply* reply)
+{
+       e->qstate->reply = reply;
+       mesh_run(mesh, e->qstate->mesh_info,
+               is_ok?module_event_reply:module_event_timeout, e);
+}
+
 struct mesh_state* 
 mesh_state_create(struct module_env* env, struct query_info* qinfo, 
        uint16_t qflags, int prime)
@@ -124,10 +187,12 @@ mesh_state_create(struct module_env* env, struct query_info* qinfo,
                region_destroy(region);
                return NULL;
        }
+       memset(mstate, 0, sizeof(*mstate));
        mstate->node = *RBTREE_NULL;
        mstate->run_node = *RBTREE_NULL;
        mstate->node.key = mstate;
        mstate->run_node.key = mstate;
+       mstate->debug_flags = 0;
        mstate->is_priming = prime;
        mstate->reply_list = NULL;
        rbtree_init(&mstate->super_set, &mesh_state_ref_compare);
@@ -173,3 +238,282 @@ mesh_state_cleanup(struct mesh_state* mstate)
        }
        region_destroy(mstate->s.region);
 }
+
+void 
+mesh_state_delete(struct mesh_state* mstate)
+{
+       struct mesh_area* mesh;
+       struct mesh_state_ref* super, ref;
+       if(!mstate)
+               return;
+       mesh = mstate->s.env->mesh;
+       mesh_detach_subs(&mstate->s);
+       if(!mstate->reply_list && mstate->super_set.count == 0) {
+               log_assert(mesh->num_detached_states > 0);
+               mesh->num_detached_states--;
+       }
+       if(mstate->reply_list) {
+               log_assert(mesh->num_reply_states > 0);
+               mesh->num_reply_states--;
+       }
+       ref.node.key = &ref;
+       ref.s = mstate;
+       RBTREE_FOR(super, struct mesh_state_ref*, &mstate->super_set) {
+               (void)rbtree_delete(&super->s->sub_set, &ref);
+       }
+       (void)rbtree_delete(&mesh->run, mstate);
+       (void)rbtree_delete(&mesh->all, mstate);
+       mesh_state_cleanup(mstate);
+}
+
+void mesh_detach_subs(struct module_qstate* qstate)
+{
+       struct mesh_area* mesh = qstate->env->mesh;
+       struct mesh_state_ref* ref, lookup;
+       struct rbnode_t* n;
+       lookup.node.key = &lookup;
+       lookup.s = qstate->mesh_info;
+       RBTREE_FOR(ref, struct mesh_state_ref*, &qstate->mesh_info->sub_set) {
+               n = rbtree_delete(&ref->s->super_set, &lookup);
+               log_assert(n != NULL); /* must have been present */
+               if(!ref->s->reply_list && ref->s->super_set.count == 0) {
+                       mesh->num_detached_states++;
+                       log_assert(mesh->num_detached_states + 
+                               mesh->num_reply_states <= mesh->all.count);
+               }
+       }
+       rbtree_init(&qstate->mesh_info->sub_set, &mesh_state_ref_compare);
+}
+
+int mesh_attach_sub(struct module_qstate* qstate, struct query_info* qinfo,
+        uint16_t qflags, int prime, struct module_qstate** newq)
+{
+       /* find it, if not, create it */
+       struct mesh_area* mesh = qstate->env->mesh;
+       struct mesh_state* sub = mesh_area_find(mesh, qinfo, qflags, prime);
+       if(!sub) {
+               struct rbnode_t* n;
+               /* create a new one */
+               sub = mesh_state_create(qstate->env, qinfo, qflags, prime);
+               if(!sub) {
+                       log_err("mesh_attach_sub: out of memory");
+                       return 0;
+               }
+               n = rbtree_insert(&mesh->all, &sub->node);
+               log_assert(n != NULL);
+               /* set detached (it is now) */
+               mesh->num_detached_states++;
+               /* set new query state to run */
+               n = rbtree_insert(&mesh->run, &sub->run_node);
+               log_assert(n != NULL);
+               *newq = &sub->s;
+       } else
+               *newq = NULL;
+       if(!mesh_state_attachment(qstate->mesh_info, sub))
+               return 0;
+       if(!sub->reply_list && sub->super_set.count == 1) {
+               /* it used to be detached, before this one got added */
+               log_assert(mesh->num_detached_states > 0);
+               mesh->num_detached_states--;
+       }
+       /* *newq will be run when inited after the current module stops */
+       return 1;
+}
+
+int mesh_state_attachment(struct mesh_state* super, struct mesh_state* sub)
+{
+       struct rbnode_t* n;
+       struct mesh_state_ref* subref; /* points to sub, inserted in super */
+       struct mesh_state_ref* superref; /* points to super, inserted in sub */
+       if( !(subref = region_alloc(super->s.region,
+               sizeof(struct mesh_state_ref))) ||
+               !(superref = region_alloc(sub->s.region,
+               sizeof(struct mesh_state_ref))) ) {
+               log_err("mesh_state_attachment: out of memory");
+               return 0;
+       }
+       superref->node.key = superref;
+       superref->s = super;
+       subref->node.key = subref;
+       subref->s = sub;
+       n = rbtree_insert(&sub->super_set, &superref->node);
+       log_assert(n != NULL);
+       n = rbtree_insert(&super->sub_set, &subref->node);
+       log_assert(n != NULL);
+       return 1;
+}
+
+/** subtract timers and the values do not overflow or become negative */
+static void
+timeval_subtract(struct timeval* d, struct timeval* end, struct timeval* start)
+{
+#ifndef S_SPLINT_S
+       d->tv_sec = end->tv_sec - start->tv_sec;
+       while(end->tv_usec < start->tv_usec) {
+               d->tv_usec += 1000000;
+               d->tv_sec--;
+       }
+       d->tv_usec = end->tv_usec - start->tv_usec;
+#endif
+}
+
+/** add timers and the values do not overflow or become negative */
+static void
+timeval_add(struct timeval* d, struct timeval* add)
+{
+#ifndef S_SPLINT_S
+       d->tv_sec += add->tv_sec;
+       d->tv_usec += add->tv_usec;
+       while(d->tv_usec > 1000000 ) {
+               d->tv_usec -= 1000000;
+               d->tv_sec++;
+       }
+#endif
+}
+
+/**
+ * Send reply to mesh reply entry
+ * @param m: mesh state to send it for.
+ * @param rcode: if not 0, error code.
+ * @param rep: reply to send (or NULL if rcode is set).
+ * @param r: reply entry
+ */
+static void
+mesh_send_reply(struct mesh_state* m, int rcode, struct reply_info* rep,
+       struct mesh_reply* r)
+{
+       struct timeval end_time;
+       /* send the reply */
+       if(rcode) {
+               error_encode(r->query_reply.c->buffer, rcode, &m->s.qinfo,
+                       r->qid, r->qflags, &r->edns);
+               comm_point_send_reply(&r->query_reply);
+       } else {
+               size_t udp_size = r->edns.udp_size;
+               r->edns.edns_version = EDNS_ADVERTISED_VERSION;
+               r->edns.udp_size = EDNS_ADVERTISED_SIZE;
+               r->edns.ext_rcode = 0;
+               r->edns.bits &= EDNS_DO;
+               if(!reply_info_answer_encode(&m->s.qinfo, rep, r->qid, 
+                       r->qflags, r->query_reply.c->buffer, 0, 1, 
+                       m->s.env->scratch, udp_size, &r->edns, 
+                       (int)(r->edns.bits & EDNS_DO))) 
+               {
+                       error_encode(r->query_reply.c->buffer, 
+                               LDNS_RCODE_SERVFAIL, &m->s.qinfo, r->qid, 
+                               r->qflags, &r->edns);
+               }
+               comm_point_send_reply(&r->query_reply);
+       }
+       /* account */
+       if(gettimeofday(&end_time, NULL) < 0) {
+               log_err("gettimeofday: %s", strerror(errno));
+               return;
+       } else {
+               struct timeval duration;
+               timeval_subtract(&duration, &end_time, &r->start_time);
+               verbose(VERB_ALGO, "query took %d s %d usec",
+                       (int)duration.tv_sec, (int)duration.tv_usec);
+               m->s.env->mesh->replies_sent++;
+               timeval_add(&m->s.env->mesh->replies_sum_wait, &duration);
+       }
+}
+
+void mesh_query_done(struct module_qstate* qstate, int rcode,
+        struct reply_info* rep)
+{
+       struct mesh_state* m = qstate->mesh_info;
+       struct mesh_reply* r;
+       log_assert(!(m->debug_flags&1)); /* not twice! */
+       m->debug_flags |= 1;
+       for(r = m->reply_list; r; r = r->next) {
+               mesh_send_reply(m, rcode, rep, r);
+       }
+}
+
+void mesh_walk_supers(struct module_qstate* qstate, int id, int rcode,
+        void (*cb)(struct module_qstate*, int, struct module_qstate*, int))
+{
+       struct mesh_state* m = qstate->mesh_info;
+       struct mesh_area* mesh = m->s.env->mesh;
+       struct mesh_state_ref* ref;
+       log_assert(!(m->debug_flags&2)); /* not twice! */
+       m->debug_flags |= 2;
+       RBTREE_FOR(ref, struct mesh_state_ref*, &qstate->mesh_info->super_set)
+       {
+               /* make super runnable */
+               (void)rbtree_insert(&mesh->run, &ref->s->run_node);
+               /* callback */
+               (*cb)(qstate, id, &ref->s->s, rcode);
+       }
+}
+
+struct mesh_state* mesh_area_find(struct mesh_area* mesh,
+       struct query_info* qinfo, uint16_t qflags, int prime)
+{
+       struct mesh_state key;
+       struct mesh_state* result;
+
+       key.node.key = &key;
+       key.is_priming = prime;
+       key.s.qinfo = *qinfo;
+       key.s.query_flags = qflags;
+       
+       result = (struct mesh_state*)rbtree_search(&mesh->all, &key);
+       return result;
+}
+
+int mesh_state_add_reply(struct mesh_state* s, struct edns_data* edns,
+        struct comm_reply* rep, uint16_t qid, uint16_t qflags)
+{
+       struct mesh_reply* r = region_alloc(s->s.region, 
+               sizeof(struct mesh_reply));
+       if(!r)
+               return 0;
+       r->query_reply = *rep;
+       r->edns = *edns;
+       r->qid = qid;
+       r->qflags = qflags;
+       if(gettimeofday(&r->start_time, NULL) < 0) {
+               log_err("addrep: gettimeofday: %s", strerror(errno));
+               memset(&r->start_time, 0, sizeof(r->start_time));
+       }
+       r->next = s->reply_list;
+       s->reply_list = r;
+       return 1;
+
+}
+
+void mesh_run(struct mesh_area* mesh, struct mesh_state* mstate,
+       enum module_ev ev, struct outbound_entry* e)
+{
+       enum module_ext_state s;
+       verbose(VERB_ALGO, "mesh_run: start");
+       while(mstate) {
+               /* run the module */
+               (*mesh->modfunc[mstate->s.curmod]->operate)
+                       (&mstate->s, ev, mstate->s.curmod, e);
+
+               /* examine results */
+               mstate->s.reply = NULL;
+               region_free_all(mstate->s.env->scratch);
+               s = mstate->s.ext_state[mstate->s.curmod];
+               verbose(VERB_ALGO, "mesh_run: %s module exit state is %s", 
+                       mesh->modfunc[mstate->s.curmod]->name, strextstate(s));
+               if(s == module_error || s == module_finished) {
+                       /* must have called _done and _supers */
+                       log_assert(mstate->debug_flags == 3);
+                       mesh_state_delete(mstate);
+               }
+
+               /* run more modules */
+               ev = module_event_pass;
+               e = NULL;
+               if(mesh->run.count > 0) {
+                       /* pop random element off the runnable tree */
+                       mstate = (struct mesh_state*)mesh->run.root->key;
+                       (void)rbtree_delete(&mesh->run, mstate);
+               } else mstate = NULL;
+       }
+       verbose(VERB_ALGO, "mesh_run: end");
+}
index c1260920fea18e9941b6ec0b9bea1acb56d25932..58f21d2fc24a9a2cdb984034222bb48be81125e5 100644 (file)
@@ -64,6 +64,8 @@ struct mesh_area {
        int num_modules;
        /** the module callbacks, array of num_modules length (ref only) */
        struct module_func_block** modfunc;
+       /** environment for new states */
+       struct module_env* env;
 
        /** set of runnable queries (mesh_state.run_node) */
        rbtree_t run;
@@ -80,6 +82,11 @@ struct mesh_area {
         * an empty set of super-states, thus are 'toplevel' or detached
         * internal opportunistic queries */
        size_t num_detached_states;
+
+       /** number of replies sent */
+       size_t replies_sent;
+       /** sum of waiting times for the replies */
+       struct timeval replies_sum_wait;
 };
 
 /**
@@ -103,6 +110,8 @@ struct mesh_state {
        struct module_qstate s;
        /** the list of replies to clients for the results */
        struct mesh_reply* reply_list;
+       /** debug flags */
+       int debug_flags;
        /** set of superstates (that want this state's result) 
         * contains struct mesh_state_ref* */
        rbtree_t super_set;
@@ -117,7 +126,7 @@ struct mesh_state {
  */
 struct mesh_state_ref {
        /** node in rbtree for set, key is this structure */
-       rbtree_t node;
+       rbnode_t node;
        /** the mesh state */
        struct mesh_state* s;
 };
@@ -147,10 +156,11 @@ struct mesh_reply {
  * @param num_modules: number of modules that are present.
  * @param modfunc: array passed (alloced and deleted by caller), that has
  *     num_modules function callbacks for the modules.
+ * @param env: environment for new queries.
  * @return mesh: the new mesh or NULL on error.
  */
 struct mesh_area* mesh_create(int num_modules, 
-       struct module_func_block** modfunc);
+       struct module_func_block** modfunc, struct module_env* env);
 
 /**
  * Delete mesh, and all query states and replies in it.
@@ -192,7 +202,8 @@ void mesh_report_reply(struct mesh_area* mesh, struct outbound_entry* e,
 /**
  * Detach-subqueries.
  * Remove all sub-query references from this query state.
- * Keeps sub-query-super-references correct.
+ * Keeps super-references of those sub-queries correct.
+ * Updates stat items in mesh_area structure.
  * @param qstate: used to find mesh state.
  */
 void mesh_detach_subs(struct module_qstate* qstate);
@@ -201,6 +212,7 @@ void mesh_detach_subs(struct module_qstate* qstate);
  * Attach subquery.
  * Creates it if it does not exist already.
  * Keeps sub and super references correct.
+ * Updates stat items in mesh_area structure.
  * Pass if it is priming query or not.
  * return:
  *     o if error (malloc) happened.
@@ -276,4 +288,58 @@ struct mesh_state* mesh_state_create(struct module_env* env,
  */
 void mesh_state_cleanup(struct mesh_state* mstate);
 
+/**
+ * Delete mesh state, cleanup and also rbtrees and so on.
+ * Will detach from all super/subnodes.
+ * @param mstate: to remove.
+ */
+void mesh_state_delete(struct mesh_state* mstate);
+
+/**
+ * Find a mesh state in the mesh area. Pass relevant flags.
+ *
+ * @param mesh: the mesh area to look in.
+ * @param qinfo: what query
+ * @param qflags: if RD bit is set or not.
+ * @param prime: if it is a priming query.
+ * @return: mesh state or NULL if not found.
+ */
+struct mesh_state* mesh_area_find(struct mesh_area* mesh, 
+       struct query_info* qinfo, uint16_t qflags, int prime);
+
+/**
+ * Setup attachment super/sub relation between super and sub mesh state.
+ * The relation must not be present when calling the function.
+ * Does not update stat items in mesh_area.
+ * @param super: super state.
+ * @param sub: sub state.
+ * @return: 0 on alloc error.
+ */
+int mesh_state_attachment(struct mesh_state* super, struct mesh_state* sub);
+
+/**
+ * Create new reply structure and attach it to a mesh state.
+ * Does not update stat items in mesh area.
+ * @param s: the mesh state.
+ * @param edns: edns data for reply (bufsize).
+ * @param rep: comm point reply info.
+ * @param qid: ID of reply.
+ * @param qflags: original query flags.
+ * @return: 0 on alloc error.
+ */
+int mesh_state_add_reply(struct mesh_state* s, struct edns_data* edns, 
+       struct comm_reply* rep, uint16_t qid, uint16_t qflags);
+
+/**
+ * Run the mesh. Run all runnable mesh states. Which can create new
+ * runnable mesh states. Until completion. Automatically called by
+ * mesh_report_reply and mesh_new_client as needed.
+ * @param mesh: mesh area.
+ * @param mstate: first mesh state to run.
+ * @param ev: event the mstate. Others get event_pass.
+ * @param e: if a reply, its outbound entry.
+ */
+void mesh_run(struct mesh_area* mesh, struct mesh_state* mstate, 
+       enum module_ev ev, struct outbound_entry* e);
+
 #endif /* SERVICES_MESH_H */
index 0add5fd43cc260a363acaeb040424374afc55394..6db5619ce8fac687cf275a75e7c4047594a9a189 100644 (file)
@@ -754,3 +754,40 @@ qinfo_query_encode(ldns_buffer* pkt, struct query_info* qinfo)
        ldns_buffer_write_u16(pkt, qinfo->qclass);
        ldns_buffer_flip(pkt);
 }
+
+void 
+error_encode(ldns_buffer* buf, int r, struct query_info* qinfo,
+       uint16_t qid, uint16_t qflags, struct edns_data* edns)
+{
+       uint16_t flags;
+
+       ldns_buffer_clear(buf);
+       ldns_buffer_write(buf, &qid, sizeof(uint16_t));
+       flags = (uint16_t)(BIT_QR | BIT_RA | r); /* QR and retcode*/
+       flags |= (qflags & (BIT_RD|BIT_CD)); /* copy RD and CD bit */
+       ldns_buffer_write_u16(buf, flags);
+       if(qinfo) flags = 1;
+       else    flags = 0;
+       ldns_buffer_write_u16(buf, flags);
+       flags = 0;
+       ldns_buffer_write(buf, &flags, sizeof(uint16_t));
+       ldns_buffer_write(buf, &flags, sizeof(uint16_t));
+       ldns_buffer_write(buf, &flags, sizeof(uint16_t));
+       if(qinfo) {
+               ldns_buffer_write(buf, qinfo->qname, qinfo->qname_len);
+               ldns_buffer_write_u16(buf, qinfo->qtype);
+               ldns_buffer_write_u16(buf, qinfo->qclass);
+       }
+       ldns_buffer_flip(buf);
+       if(edns) {
+               struct edns_data es = *edns;
+               es.edns_version = EDNS_ADVERTISED_VERSION;
+               es.udp_size = EDNS_ADVERTISED_SIZE;
+               es.ext_rcode = 0;
+               es.bits &= EDNS_DO;
+               if(ldns_buffer_limit(buf) + calc_edns_field_size(&es) >
+                       edns->udp_size)
+                       return;
+               attach_edns_record(buf, &es);
+       }
+}
index 203929dc832c56d1ab2339f51141969aa4bde9ef..11fae6be11abfeea3f5604f0b28638741aaeaacd 100644 (file)
@@ -112,5 +112,18 @@ uint16_t calc_edns_field_size(struct edns_data* edns);
  */
 void attach_edns_record(ldns_buffer* pkt, struct edns_data* edns);
 
+/** 
+ * Encode an error. With QR and RA set.
+ *
+ * @param pkt: where to store the packet.
+ * @param r: RCODE value to encode.
+ * @param qinfo: if not NULL, the query is included.
+ * @param qid: query ID to set in packet. network order.
+ * @param qflags: original query flags (to copy RD and CD bits). host order.
+ * @param edns: if not NULL, this is the query edns info,
+ *     and an edns reply is attached. Only attached if EDNS record fits reply.
+ */
+void error_encode(ldns_buffer* pkt, int r, struct query_info* qinfo,
+       uint16_t qid, uint16_t qflags, struct edns_data* edns);
 
 #endif /* UTIL_DATA_MSGENCODE_H */
index f10b97bef0c1ed7250d5527ff23bdabb27c175a6..52ceaa7cbd35be35aaafdb6de765e77bbd320112 100644 (file)
@@ -120,14 +120,75 @@ struct module_env {
                struct module_qstate* q);
 
        /**
-        * Cleanup subqueries from this query state. Either delete or
-        * move them somewhere else. This query state no longer needs the
-        * results from those subqueries.
-        * @param qstate: query state.
-        *      subqueries are (re)moved so that no subq_done events from
-        *      them will reach this qstate.
+        * Detach-subqueries.
+        * Remove all sub-query references from this query state.
+        * Keeps super-references of those sub-queries correct.
+        * Updates stat items in mesh_area structure.
+        * @param qstate: used to find mesh state.
         */
-       void (*remove_subqueries)(struct module_qstate* qstate);
+       void (*detach_subs)(struct module_qstate* qstate);
+
+       /**
+        * Attach subquery.
+        * Creates it if it does not exist already.
+        * Keeps sub and super references correct.
+        * Updates stat items in mesh_area structure.
+        * Pass if it is priming query or not.
+        * return:
+        * o if error (malloc) happened.
+        * o need to initialise the new state (module init; it is a new state).
+        *   so that the next run of the query with this module is successful.
+        * o no init needed, attachment successful.
+        * 
+        * @param qstate: the state to find mesh state, and that wants to 
+        *      receive the results from the new subquery.
+        * @param qinfo: what to query for (copied).
+        * @param qflags: what flags to use (RD flag or not).
+        * @param prime: if it is a (stub) priming query.
+        * @param newq: If the new subquery needs initialisation, it is 
+        *      returned, otherwise NULL is returned.
+        * @return: false on error, true if success (and init may be needed).
+        */ 
+       int (*attach_sub)(struct module_qstate* qstate, 
+               struct query_info* qinfo, uint16_t qflags, int prime, 
+               struct module_qstate** newq);
+
+       /**
+        * Query state is done, send messages to reply entries.
+        * Encode messages using reply entry values and the querystate 
+        * (with original qinfo), using given reply_info.
+        * Pass errcode != 0 if an error reply is needed.
+        * If no reply entries, nothing is done.
+        * Must be called before a module can module_finished or return 
+        * module_error.
+        * The module must handle the super query states itself as well.
+        * 
+        * @param qstate: used for original query info. And to find mesh info.
+        * @param rcode: if not 0 (NOERROR) an error is sent back (and 
+        *      rep ignored).
+        * @param rep: reply to encode and send back to clients.
+        */
+       void (*query_done)(struct module_qstate* qstate, int rcode,
+               struct reply_info* rep);
+
+       /**
+        * Get a callback for the super query states that are interested in the 
+        * results from this query state. These can then be changed for error 
+        * or results.
+        * Must be called befor a module can module_finished or return 
+        * module_error.  After finishing or module error, the super 
+        * query states become runnable with event module_event_pass.
+        * 
+        * @param qstate: the state that has results, used to find mesh state.
+        * @param id: module id.
+        * @param rcode: rcode to pass to callback, for easier error passing to 
+        *       parents.
+        * @param cb: callback function. Called as
+        *      cb(qstate, id, super_qstate, rcode) for every super qstate.
+        */
+       void (*walk_supers)(struct module_qstate* qstate, int id, 
+               int rcode, void (*cb)(struct module_qstate*, int, 
+               struct module_qstate*, int));
 
        /** region for temporary usage. May be cleared after operate() call. */
        struct region* scratch;
@@ -141,6 +202,17 @@ struct module_env {
        struct ub_randstate* rnd;
        /** module specific data. indexed by module id. */
        void* modinfo[MAX_MODULE];
+
+       /** @@@ TO BE DELETED */
+       /**
+        * Cleanup subqueries from this query state. Either delete or
+        * move them somewhere else. This query state no longer needs the
+        * results from those subqueries.
+        * @param qstate: query state.
+        *      subqueries are (re)moved so that no subq_done events from
+        *      them will reach this qstate.
+        */
+       void (*remove_subqueries)(struct module_qstate* qstate);
 };
 
 /**