dq.c File Reference

---

Detailed Description

Dynamic querying.

Author:

Raphael Manfredi

Date:

2004

#include "common.h"
#include <math.h>
#include "dq.h"
#include "mq.h"
#include "gmsg.h"
#include "pmsg.h"
#include "nodes.h"
#include "gnet_stats.h"
#include "qrp.h"
#include "vmsg.h"
#include "search.h"
#include "alive.h"
#include "if/gnet_property_priv.h"
#include "lib/atoms.h"
#include "lib/cq.h"
#include "lib/endian.h"
#include "lib/glib-missing.h"
#include "lib/misc.h"
#include "lib/tm.h"
#include "lib/walloc.h"
#include "lib/override.h"

Data Structures
struct	next_up
	Structure produced by dq_fill_next_up, representing the nodes to which we could send the query, along with routing information to be able to favor UPs that report a QRP match early in the querying process. More...
struct	dquery
	The dynamic query. More...
struct	pmsg_info
	Information about query messages sent. More...
struct	cancel_context
Defines
#define	DQ_MAX_LIFETIME   300000 /**< 5 minutes, in ms */
	5 minutes, in ms
#define	DQ_PROBE_TIMEOUT   1500 /**< 1.5 s extra per connection */
	1.5 s extra per connection
#define	DQ_PENDING_TIMEOUT   1200 /**< 1.2 s extra per pending message */
	1.2 s extra per pending message
#define	DQ_QUERY_TIMEOUT   3700 /**< 3.7 s */
	3.7 s
#define	DQ_TIMEOUT_ADJUST   100 /**< 100 ms at each connection */
	100 ms at each connection
#define	DQ_MIN_TIMEOUT   1500 /**< 1.5 s at least between queries */
	1.5 s at least between queries
#define	DQ_LINGER_TIMEOUT   120000 /**< 2 minutes, in ms */
	2 minutes, in ms
#define	DQ_STATUS_TIMEOUT   30000 /**< 30 s, in ms, to reply to query status */
	30 s, in ms, to reply to query status
#define	DQ_MAX_PENDING   3 /**< Max pending queries we allow */
	Max pending queries we allow.
#define	DQ_LEAF_RESULTS   50 /**< # of results targetted for leaves */
	# of results targetted for leaves
#define	DQ_LOCAL_RESULTS   150 /**< # of results for local queries */
	# of results for local queries
#define	DQ_SHA1_DECIMATOR   25 /**< Divide expected by that much for SHA1 */
	Divide expected by that much for SHA1.
#define	DQ_PROBE_UP   3 /**< Amount of UPs for initial probe */
	Amount of UPs for initial probe.
#define	DQ_MAX_HORIZON   500000 /**< Stop after that many UP queried */
	Stop after that many UP queried.
#define	DQ_MIN_HORIZON   3000 /**< Min horizon before timeout adjustment */
	Min horizon before timeout adjustment.
#define	DQ_LOW_RESULTS   10 /**< After DQ_MIN_HORIZON queried for adj. */
	After DQ_MIN_HORIZON queried for adj.
#define	DQ_PERCENT_KEPT   5 /**< Assume 5% of results kept, worst case */
	Assume 5% of results kept, worst case.
#define	DQ_MAX_TTL   5 /**< Max TTL we can use */
	Max TTL we can use.
#define	DQ_AVG_ULTRA_NODES   2 /**< Avg # of ultranodes a leaf queries */
	Avg # of ultranodes a leaf queries.
#define	DQ_MQ_EPSILON   2048 /**< Queues identical at +/- 2K */
	Queues identical at +/- 2K.
#define	DQ_FUZZY_FACTOR   0.80 /**< Corrector for theoretical horizon */
	Corrector for theoretical horizon.
#define	QUERY_TEXT(m)   ((m) + sizeof(struct gnutella_header) + 2)
	Compute start of search string (which is NUL terminated) in query.
#define	DQ_F_ID_CLEANING   0x00000001 /**< Cleaning the `by_node_id' table */
	Cleaning the `by_node_id' table.
#define	DQ_F_LINGER   0x00000002 /**< Lingering to monitor extra hits */
	Lingering to monitor extra hits.
#define	DQ_F_LEAF_GUIDED   0x00000004 /**< Leaf-guided query */
	Leaf-guided query.
#define	DQ_F_WAITING   0x00000008 /**< Waiting guidance reply from leaf */
	Waiting guidance reply from leaf.
#define	DQ_F_GOT_GUIDANCE   0x00000010 /**< Got unsolicited leaf guidance */
	Got unsolicited leaf guidance.
#define	DQ_F_USR_CANCELLED   0x00000020 /**< Explicitely cancelled by user */
	Explicitely cancelled by user.
#define	DQ_F_EXITING   0x80000000 /**< Final cleanup at exit time */
	Final cleanup at exit time.
#define	MAX_DEGREE   50
#define	MAX_TTL   5
Typedefs
typedef dquery	dquery_t
	The dynamic query.
Functions
	RCSID ("$Id:dq.c, v 1.31 2005/09/28 21:41:37 rmanfredi Exp $")
void	dq_send_next (dquery_t *dq)
	Iterate over the UPs which have not seen our query yet, select one and send it the query.
void	dq_terminate (dquery_t *dq)
	Terminate active querying.
void	fill_hosts (void)
	Compute the hosts[] table so that:.
guint32	dq_get_horizon (gint degree, gint ttl)
	Computes theoretical horizon reached by a query sent to a host advertising a given degree if it is going to travel ttl hops.
guint32	dq_kept_results (dquery_t *dq)
	Compute amount of results "kept" for the query, if we have this information available.
gint	dq_select_ttl (dquery_t *dq, gnutella_node_t *node, gint connections)
	Select the proper TTL for the next query we're going to send to the specified node, assuming hosts are equally split among the remaining connections we have yet to query.
pmsg_info *	dq_pmi_alloc (dquery_t *dq, guint16 degree, guint8 ttl, guint32 node_id)
	Create a pmsg_info structure, giving meta-information about the message we're about to send.
void	dq_pmi_free (struct pmsg_info *pmi)
	Get rid of the pmsg_info structure.
gboolean	dq_alive (dquery_t *dq, guint32 qid)
	Check whether query bearing the specified ID is still alive and has not been cancelled yet.
void	dq_pmsg_free (pmsg_t *mb, gpointer arg)
	Free routine for an extended message block.
pmsg_t *	dq_pmsg_by_ttl (dquery_t *dq, gint ttl)
	Fetch message for a given TTL.
gint	dq_fill_probe_up (dquery_t *dq, gnutella_node_t **nv, gint ncount)
	Fill node vector with UP hosts to which we could send our probe query.
gint	dq_fill_next_up (dquery_t *dq, struct next_up *nv, gint ncount)
	Fill node vector with UP hosts to which we could send our next query.
void	dq_sendto_leaves (dquery_t *dq, gnutella_node_t *source)
	Forward message to all the leaves but the one originating this query, according to their QRP tables.
void	dq_free (dquery_t *dq)
	Release the dynamic query object.
void	dq_expired (cqueue_t *unused_cq, gpointer obj)
	Callout queue callback invoked when the dynamic query has expired.
void	dq_results_expired (cqueue_t *unused_cq, gpointer obj)
	Callout queue callback invoked when the result timer has expired.
gint	node_mq_cmp (const void *np1, const void *np2)
	qsort() callback for sorting nodes by increasing queue size.
gint	node_mq_qrp_cmp (const void *np1, const void *np2)
	qsort() callback for sorting nodes by increasing queue size, with a preference towards nodes that have a QRP match.
void	dq_send_query (dquery_t *dq, gnutella_node_t *n, gint ttl)
	Send individual query to selected node at the supplied TTL.
void	dq_send_probe (dquery_t *dq)
	Send probe query (initial querying).
void	dq_common_init (dquery_t *dq)
	Common initialization code for a dynamic query.
void	dq_launch_net (gnutella_node_t *n, query_hashvec_t *qhv)
	Start new dynamic query out of a message we got from one of our leaves.
void	dq_launch_local (gnet_search_t handle, pmsg_t *mb, query_hashvec_t *qhv)
	Start new dynamic query for a local search.
void	dq_node_removed (guint32 node_id)
	Tells us a node ID has been removed.
gboolean	dq_count_results (gchar *muid, gint count, gboolean oob)
	Common code to count the results.
gboolean	dq_got_results (gchar *muid, guint count)
	Called every time we successfully parsed a query hit from the network.
gboolean	dq_oob_results_ind (gchar *muid, gint count)
	Called every time we get notified about the presence of some OOB hits.
void	dq_oob_results_got (const gchar *muid, guint count)
	Called when OOB results were received, after dq_got_results() was called to record them.
void	dq_got_query_status (gchar *muid, guint32 node_id, guint16 kept)
	Called when we get a "Query Status Response" message where the querying node informs us about the amount of results he kept after filtering.
void	dq_cancel_local (gpointer key, gpointer unused_value, gpointer udata)
	Cancel local query bearing the specified search handle.
void	dq_search_closed (gnet_search_t handle)
	Invoked when a local search is closed.
gboolean	dq_get_results_wanted (gchar *muid, guint32 *wanted)
	Called for OOB-proxied queries when we get an "OOB Reply Indication" from remote hosts.
void	dq_init (void)
	Initialize dynamic querying.
void	free_query (gpointer key, gpointer unused_value, gpointer unused_udata)
	Hashtable iteration callback to free the dquery_t object held as the key.
void	free_query_list (gpointer key, gpointer value, gpointer unused_udata)
	Hashtable iteration callback to free the items remaining in the by_node_id table.
void	free_muid (gpointer key, gpointer unused_value, gpointer unused_udata)
	Hashtable iteration callback to free the MUIDs in the `by_muid' table.
void	dq_close (void)
	Cleanup data structures used by dynamic querying.
Variables
GHashTable *	dqueries = NULL
	This table keeps track of all the dynamic query objects that we have created and which are alive.
GHashTable *	by_node_id = NULL
	This table keeps track of all the dynamic query objects created for a given node ID.
GHashTable *	by_muid = NULL
	This table keeps track of the association between a MUID and the dynamic query, so that when results come back, we may account them for the relevant query.
guint32	hosts [MAX_DEGREE][MAX_TTL]
	Pre-computed horizon.
guint32	dyn_query_id = 0

---

Define Documentation

#define DQ_AVG_ULTRA_NODES   2 /**< Avg # of ultranodes a leaf queries */

Avg # of ultranodes a leaf queries.

#define DQ_F_EXITING   0x80000000 /**< Final cleanup at exit time */

Final cleanup at exit time.

#define DQ_F_GOT_GUIDANCE   0x00000010 /**< Got unsolicited leaf guidance */

Got unsolicited leaf guidance.

#define DQ_F_ID_CLEANING   0x00000001 /**< Cleaning the `by_node_id' table */

Cleaning the `by_node_id' table.

#define DQ_F_LEAF_GUIDED   0x00000004 /**< Leaf-guided query */

Leaf-guided query.

#define DQ_F_LINGER   0x00000002 /**< Lingering to monitor extra hits */

Lingering to monitor extra hits.

#define DQ_F_USR_CANCELLED   0x00000020 /**< Explicitely cancelled by user */

Explicitely cancelled by user.

#define DQ_F_WAITING   0x00000008 /**< Waiting guidance reply from leaf */

Waiting guidance reply from leaf.

#define DQ_FUZZY_FACTOR   0.80 /**< Corrector for theoretical horizon */

Corrector for theoretical horizon.

#define DQ_LEAF_RESULTS   50 /**< # of results targetted for leaves */

# of results targetted for leaves

#define DQ_LINGER_TIMEOUT   120000 /**< 2 minutes, in ms */

2 minutes, in ms

#define DQ_LOCAL_RESULTS   150 /**< # of results for local queries */

# of results for local queries

#define DQ_LOW_RESULTS   10 /**< After DQ_MIN_HORIZON queried for adj. */

After DQ_MIN_HORIZON queried for adj.

#define DQ_MAX_HORIZON   500000 /**< Stop after that many UP queried */

Stop after that many UP queried.

#define DQ_MAX_LIFETIME   300000 /**< 5 minutes, in ms */

5 minutes, in ms

#define DQ_MAX_PENDING   3 /**< Max pending queries we allow */

Max pending queries we allow.

#define DQ_MAX_TTL   5 /**< Max TTL we can use */

Max TTL we can use.

#define DQ_MIN_HORIZON   3000 /**< Min horizon before timeout adjustment */

Min horizon before timeout adjustment.

#define DQ_MIN_TIMEOUT   1500 /**< 1.5 s at least between queries */

1.5 s at least between queries

#define DQ_MQ_EPSILON   2048 /**< Queues identical at +/- 2K */

Queues identical at +/- 2K.

#define DQ_PENDING_TIMEOUT   1200 /**< 1.2 s extra per pending message */

1.2 s extra per pending message

#define DQ_PERCENT_KEPT   5 /**< Assume 5% of results kept, worst case */

Assume 5% of results kept, worst case.

#define DQ_PROBE_TIMEOUT   1500 /**< 1.5 s extra per connection */

1.5 s extra per connection

#define DQ_PROBE_UP   3 /**< Amount of UPs for initial probe */

Amount of UPs for initial probe.

#define DQ_QUERY_TIMEOUT   3700 /**< 3.7 s */

3.7 s

#define DQ_SHA1_DECIMATOR   25 /**< Divide expected by that much for SHA1 */

Divide expected by that much for SHA1.

#define DQ_STATUS_TIMEOUT   30000 /**< 30 s, in ms, to reply to query status */

30 s, in ms, to reply to query status

#define DQ_TIMEOUT_ADJUST   100 /**< 100 ms at each connection */

100 ms at each connection

#define MAX_DEGREE   50

#define MAX_TTL   5

#define QUERY_TEXT (  m   )     ((m) + sizeof(struct gnutella_header) + 2)

Compute start of search string (which is NUL terminated) in query. The "+2" skips the "speed" field in the query.

---

Typedef Documentation

typedef struct dquery dquery_t

The dynamic query.

---

Function Documentation

gboolean dq_alive (  dquery_t *  dq, guint32  qid )  [static]

Check whether query bearing the specified ID is still alive and has not been cancelled yet.

void dq_cancel_local (  gpointer  key, gpointer  unused_value, gpointer  udata )  [static]

Cancel local query bearing the specified search handle. -- hash table iterator callback

void dq_close (  void   )

Cleanup data structures used by dynamic querying.

void dq_common_init (  dquery_t *  dq  )  [static]

Common initialization code for a dynamic query.

gboolean dq_count_results (  gchar *  muid, gint  count, gboolean  oob )  [static]

Common code to count the results. Parameters: muid  is the dynamic query's MUID, i.e. the MUID used to send out the query on the network (important for OOB-proxied queries). count  is the amount of results we received or got notified about oob  if TRUE indicates that we just got notified about OOB results awaiting, but which have not been claimed yet. If FALSE, the results have been validated and will be sent to the queryier. Returns: FALSE if the query was explicitly cancelled by the user

void dq_expired (  cqueue_t *  unused_cq, gpointer  obj )  [static]

Callout queue callback invoked when the dynamic query has expired.

gint dq_fill_next_up (  dquery_t *  dq, struct next_up *  nv, gint  ncount )  [static]

Fill node vector with UP hosts to which we could send our next query. Parameters: dq  the dynamic query nv  the pre-allocated new node vector ncount  the size of the vector Returns: amount of nodes we found.

gint dq_fill_probe_up (  dquery_t *  dq, gnutella_node_t **  nv, gint  ncount )  [static]

Fill node vector with UP hosts to which we could send our probe query. Parameters: dq  the dynamic query nv  the pre-allocated node vector ncount  the size of the vector Returns: amount of nodes we found.

void dq_free (  dquery_t *  dq  )  [static]

Release the dynamic query object.

guint32 dq_get_horizon (  gint  degree, gint  ttl )  [static]

Computes theoretical horizon reached by a query sent to a host advertising a given degree if it is going to travel ttl hops. We adjust the horizon by DQ_FUZZY_FACTOR, assuming that at each hop there is deperdition due to flow-control, network cycles, etc...

gboolean dq_get_results_wanted (  gchar *  muid, guint32 *  wanted )

Called for OOB-proxied queries when we get an "OOB Reply Indication" from remote hosts. The aim is to determine whether the query still needs results, to decide whether we'll claim the advertised results or not. Parameters: muid  the message ID of the query wanted  where the amount of results still expected is written Returns: TRUE if the query is still active, FALSE if it does not exist any more, in which case nothing is returned into `wanted'.

void dq_got_query_status (  gchar *  muid, guint32  node_id, guint16  kept )

Called when we get a "Query Status Response" message where the querying node informs us about the amount of results he kept after filtering. Parameters: muid  is the search MUID. node_id  is the ID of the node that sent us the status response. we check that it is the one for the query, to avoid a neighbour telling us about a search it did not issue! kept  is the amount of results they kept. The special value 0xffff is a request to stop the query immediately.

gboolean dq_got_results (  gchar *  muid, guint  count )

Called every time we successfully parsed a query hit from the network. If we have a dynamic query registered for the MUID, increase the result count. Returns: FALSE if the query was explicitly cancelled by the user and results should be dropped, TRUE otherwise. In other words, returns whether we should forward the results.

void dq_init (  void   )

Initialize dynamic querying.

guint32 dq_kept_results (  dquery_t *  dq  )  [static]

Compute amount of results "kept" for the query, if we have this information available.

void dq_launch_local (  gnet_search_t  handle, pmsg_t *  mb, query_hashvec_t *  qhv )

Start new dynamic query for a local search. We become the owner of the `mb' and `qhv' pointers.

void dq_launch_net (  gnutella_node_t *  n, query_hashvec_t *  qhv )

Start new dynamic query out of a message we got from one of our leaves.

void dq_node_removed (  guint32  node_id  )

Tells us a node ID has been removed. Get rid of all the queries registered for that node.

void dq_oob_results_got (  const gchar *  muid, guint  count )

Called when OOB results were received, after dq_got_results() was called to record them. We need to undo the accounting made when dq_oob_results_ind() was called (to register unclaimed hits, which were finally claimed and parsed).

gboolean dq_oob_results_ind (  gchar *  muid, gint  count )

Called every time we get notified about the presence of some OOB hits. The hits have not yet been claimed. Returns: FALSE if the query was explicitly cancelled by the user and results should not be claimed.

struct pmsg_info* dq_pmi_alloc (  dquery_t *  dq, guint16  degree, guint8  ttl, guint32  node_id )  [static]

Create a pmsg_info structure, giving meta-information about the message we're about to send. Parameters: dq  the dynamic query degree  the degree of the node to which the message is sent ttl  the TTL at which the message is sent node_id  the ID of the node to which we send the message

void dq_pmi_free (  struct pmsg_info *  pmi  )  [static]

Get rid of the pmsg_info structure.

pmsg_t* dq_pmsg_by_ttl (  dquery_t *  dq, gint  ttl )  [static]

Fetch message for a given TTL. If no such message exists yet, create it from the "template" message.

void dq_pmsg_free (  pmsg_t *  mb, gpointer  arg )  [static]

Free routine for an extended message block.

void dq_results_expired (  cqueue_t *  unused_cq, gpointer  obj )  [static]

Callout queue callback invoked when the result timer has expired.

void dq_search_closed (  gnet_search_t  handle  )

Invoked when a local search is closed.

gint dq_select_ttl (  dquery_t *  dq, gnutella_node_t *  node, gint  connections )  [static]

Select the proper TTL for the next query we're going to send to the specified node, assuming hosts are equally split among the remaining connections we have yet to query.

void dq_send_next (  dquery_t *  dq  )  [static]

Iterate over the UPs which have not seen our query yet, select one and send it the query. If no more UP remain, terminate this query.

void dq_send_probe (  dquery_t *  dq  )  [static]

Send probe query (initial querying). This can generate up to DQ_PROBE_UP individual queries.

void dq_send_query (  dquery_t *  dq, gnutella_node_t *  n, gint  ttl )  [static]

Send individual query to selected node at the supplied TTL. If the node advertises a lower maximum TTL, the supplied TTL is adjusted down accordingly.

void dq_sendto_leaves (  dquery_t *  dq, gnutella_node_t *  source )  [static]

Forward message to all the leaves but the one originating this query, according to their QRP tables. Attention: NB: In order to avoid qrt_build_query_target() selecting neighbouring ultra nodes that support last-hop QRP, we ensure the TTL is NOT 1. This is why we somehow duplicate qrt_route_query() here.

void dq_terminate (  dquery_t *  dq  )  [static]

Terminate active querying.

void fill_hosts (  void   )  [static]

Compute the hosts[] table so that:. hosts[i][j] = Sum[i^k, 0 <= k <= j] following the formula: hosts(degree,ttl) = Sum[(degree-1)^i, 0 <= i <= ttl-1]

void free_muid (  gpointer  key, gpointer  unused_value, gpointer  unused_udata )  [static]

Hashtable iteration callback to free the MUIDs in the `by_muid' table. Normally, after having freed the dqueries table, there should not be anything remaining, hence warn!

void free_query (  gpointer  key, gpointer  unused_value, gpointer  unused_udata )  [static]

Hashtable iteration callback to free the dquery_t object held as the key.

void free_query_list (  gpointer  key, gpointer  value, gpointer  unused_udata )  [static]

Hashtable iteration callback to free the items remaining in the by_node_id table. Normally, after having freed the dqueries table, there should not be anything remaining, hence warn!

gint node_mq_cmp (  const void *  np1, const void *  np2 )  [static]

qsort() callback for sorting nodes by increasing queue size.

gint node_mq_qrp_cmp (  const void *  np1, const void *  np2 )  [static]

qsort() callback for sorting nodes by increasing queue size, with a preference towards nodes that have a QRP match.

RCSID (  "$Id:dq.  c, v 1.31 2005/09/28 21:41:37 rmanfredi Exp $"  )

---

Variable Documentation

GHashTable* by_muid = NULL [static]

This table keeps track of the association between a MUID and the dynamic query, so that when results come back, we may account them for the relevant query. The keys are MUIDs (GUID atoms), the values are the dquery_t object.

GHashTable* by_node_id = NULL [static]

This table keeps track of all the dynamic query objects created for a given node ID. The key is the node ID (converted to a pointer) and the value is a GSList containing all the queries for that node.

GHashTable* dqueries = NULL [static]

This table keeps track of all the dynamic query objects that we have created and which are alive.

guint32 dyn_query_id = 0 [static]

guint32 hosts[MAX_DEGREE][MAX_TTL] [static]

Pre-computed horizon.

---