resolver.c

基于sip协议的网络电话源码

    if (status != PJ_SUCCESS) {
	cache = pj_hash_get(resolver->hrescache, key, sizeof(*key), &hval);
	if (cache)
	    pj_list_push_back(&resolver->res_free_nodes, cache);
	pj_hash_set(NULL, resolver->hrescache, key, sizeof(*key), hval, NULL);
    }


    /* Calculate expiration time. */
    if (set_expiry) {
	if (pkt->hdr.anscount == 0 || status != PJ_SUCCESS) {
	    /* If we don't have answers for the name, then give a different
	     * ttl value (note: PJ_DNS_RESOLVER_INVALID_TTL may be zero, 
	     * which means that invalid names won't be kept in the cache)
	     */
	    ttl = PJ_DNS_RESOLVER_INVALID_TTL;

	} else {
	    /* Otherwise get the minimum TTL from the answers */
	    unsigned i;
	    ttl = 0xFFFFFFFF;
	    for (i=0; i<pkt->hdr.anscount; ++i) {
		if (pkt->ans[i].ttl < ttl)
		    ttl = pkt->ans[i].ttl;
	    }
	}
    } else {
	ttl = 0xFFFFFFFF;
    }

    /* Apply maximum TTL */
    if (ttl > resolver->settings.cache_max_ttl)
	ttl = resolver->settings.cache_max_ttl;

    /* If TTL is zero, clear the same entry in the hash table */
    if (ttl == 0) {
	cache = pj_hash_get(resolver->hrescache, key, sizeof(*key), &hval);
	if (cache)
	    pj_list_push_back(&resolver->res_free_nodes, cache);
	pj_hash_set(NULL, resolver->hrescache, key, sizeof(*key), hval, NULL);
	return;
    }

    /* Get a cache response entry */
    cache = pj_hash_get(resolver->hrescache, key, sizeof(*key), &hval);
    if (cache == NULL) {
	if (!pj_list_empty(&resolver->res_free_nodes)) {
	    cache = resolver->res_free_nodes.next;
	    pj_list_erase(cache);
	} else {
	    cache = pj_pool_zalloc(resolver->pool, sizeof(*cache));
	}
    }

    /* Duplicate the packet.
     * We don't need to keep the query, NS, and AR sections from the packet,
     * so exclude from duplication.
     */
    res_pool = pj_pool_create_on_buf("respool", cache->buf, sizeof(cache->buf));
    PJ_TRY {
	cache->pkt = NULL;
	pj_dns_packet_dup(res_pool, pkt, 
			  PJ_DNS_NO_QD | PJ_DNS_NO_NS | PJ_DNS_NO_AR,
			  &cache->pkt);
    }
    PJ_CATCH_ANY {
	PJ_LOG(1,(THIS_FILE, 
		  "Not enough memory to duplicate DNS response. Response was "
		  "truncated."));
    }
    PJ_END;

    /* Calculate expiration time */
    if (set_expiry) {
	pj_gettimeofday(&cache->expiry_time);
	cache->expiry_time.sec += ttl;
    } else {
	cache->expiry_time.sec = 0x7FFFFFFFL;
	cache->expiry_time.msec = 0;
    }

    /* Copy key to the cached response */
    pj_memcpy(&cache->key, key, sizeof(*key));

    /* Update the hash table */
    pj_hash_set_np(resolver->hrescache, &cache->key, sizeof(*key), hval,
		   cache->hbuf, cache);
}


/* Callback to be called when query has timed out */
static void on_timeout( pj_timer_heap_t *timer_heap,
			struct pj_timer_entry *entry)
{
    pj_dns_resolver *resolver;
    pj_dns_async_query *q, *cq;
    pj_status_t status;

    PJ_UNUSED_ARG(timer_heap);

    q = entry->user_data;
    resolver = q->resolver;

    pj_mutex_lock(resolver->mutex);

    /* Recheck that this query is still pending, since there is a slight
     * possibility of race condition (timer elapsed while at the same time
     * response arrives)
     */
    if (pj_hash_get(resolver->hquerybyid, &q->id, sizeof(q->id), NULL)==NULL) {
	/* Yeah, this query is done. */
	pj_mutex_unlock(resolver->mutex);
	return;
    }

    /* Invalidate id. */
    q->timer_entry.id = 0;

    /* Check to see if we should retransmit instead of time out */
    if (q->transmit_cnt < PJ_DNS_RESOLVER_QUERY_RETRANSMIT_COUNT) {
	status = transmit_query(resolver, q);
	if (status == PJ_SUCCESS) {
	    pj_mutex_unlock(resolver->mutex);
	    return;
	} else {
	    /* Error occurs */
	    char errmsg[PJ_ERR_MSG_SIZE];

	    pj_strerror(status, errmsg, sizeof(errmsg));
	    PJ_LOG(4,(resolver->name.ptr,
		      "Error transmitting request: %s", errmsg));

	    /* Let it fallback to timeout section below */
	}
    }

    /* Clear hash table entries */
    pj_hash_set(NULL, resolver->hquerybyid, &q->id, sizeof(q->id), 0, NULL);
    pj_hash_set(NULL, resolver->hquerybyres, &q->key, sizeof(q->key), 0, NULL);

    /* Call application callback, if any. */
    if (q->cb)
	(*q->cb)(q->user_data, PJ_ETIMEDOUT, NULL);

    /* Call application callback for child queries. */
    cq = q->child_head.next;
    while (cq != (void*)&q->child_head) {
	if (cq->cb)
	    (*cq->cb)(cq->user_data, PJ_ETIMEDOUT, NULL);
	cq = cq->next;
    }

    /* Clear data */
    q->timer_entry.id = 0;
    q->user_data = NULL;

    /* Put child entries into recycle list */
    cq = q->child_head.next;
    while (cq != (void*)&q->child_head) {
	pj_dns_async_query *next = cq->next;
	pj_list_push_back(&resolver->query_free_nodes, cq);
	cq = next;
    }

    /* Put query entry into recycle list */
    pj_list_push_back(&resolver->query_free_nodes, q);

    pj_mutex_unlock(resolver->mutex);
}


/* Callback from ioqueue when packet is received */
static void on_read_complete(pj_ioqueue_key_t *key, 
                             pj_ioqueue_op_key_t *op_key, 
                             pj_ssize_t bytes_read)
{
    pj_dns_resolver *resolver;
    pj_pool_t *pool;
    pj_dns_parsed_packet *dns_pkt;
    pj_dns_async_query *q;
    pj_status_t status;
    PJ_USE_EXCEPTION;


    resolver = pj_ioqueue_get_user_data(key);
    pj_mutex_lock(resolver->mutex);


    /* Check for errors */
    if (bytes_read < 0) {
	char errmsg[PJ_ERR_MSG_SIZE];

	status = -bytes_read;
	pj_strerror(status, errmsg, sizeof(errmsg));
	PJ_LOG(4,(resolver->name.ptr, 
		  "DNS resolver read error from %s:%d: %s", 
		  pj_inet_ntoa(resolver->udp_src_addr.sin_addr),
		  pj_ntohs(resolver->udp_src_addr.sin_port),
		  errmsg));

	goto read_next_packet;
    }

    PJ_LOG(5,(resolver->name.ptr, 
	      "Received %d bytes DNS response from %s:%d",
	      (int)bytes_read, 
	      pj_inet_ntoa(resolver->udp_src_addr.sin_addr),
	      pj_ntohs(resolver->udp_src_addr.sin_port)));


    /* Check for zero packet */
    if (bytes_read == 0)
	goto read_next_packet;

    /* Create temporary pool from a fixed buffer */
    pool = pj_pool_create_on_buf("restmp", resolver->tmp_pool, 
				 sizeof(resolver->tmp_pool));

    /* Parse DNS response */
    status = -1;
    dns_pkt = NULL;
    PJ_TRY {
	status = pj_dns_parse_packet(pool, resolver->udp_rx_pkt, 
				     (unsigned)bytes_read, &dns_pkt);
    }
    PJ_CATCH_ANY {
	status = PJ_ENOMEM;
    }
    PJ_END;

    /* Update nameserver status */
    report_nameserver_status(resolver, &resolver->udp_src_addr, dns_pkt);

    /* Handle parse error */
    if (status != PJ_SUCCESS) {
	char errmsg[PJ_ERR_MSG_SIZE];

	pj_strerror(status, errmsg, sizeof(errmsg));
	PJ_LOG(3,(resolver->name.ptr, 
		  "Error parsing DNS response from %s:%d: %s", 
		  pj_inet_ntoa(resolver->udp_src_addr.sin_addr), 
		  pj_ntohs(resolver->udp_src_addr.sin_port), 
		  errmsg));
	goto read_next_packet;
    }

    /* Find the query based on the transaction ID */
    q = pj_hash_get(resolver->hquerybyid, &dns_pkt->hdr.id, 
		    sizeof(dns_pkt->hdr.id), NULL);
    if (!q) {
	PJ_LOG(5,(resolver->name.ptr, 
		  "Unable to find query for DNS response id=%d from %s:%d "
		  "(the query may had been answered by other name servers)",
		  (unsigned)dns_pkt->hdr.id,
		  pj_inet_ntoa(resolver->udp_src_addr.sin_addr), 
		  pj_ntohs(resolver->udp_src_addr.sin_port)));
	goto read_next_packet;
    }

    /* Map DNS Rcode in the response into PJLIB status name space */
    status = PJ_STATUS_FROM_DNS_RCODE(PJ_DNS_GET_RCODE(dns_pkt->hdr.flags));

    /* Cancel query timeout timer. */
    pj_assert(q->timer_entry.id != 0);
    pj_timer_heap_cancel(resolver->timer, &q->timer_entry);
    q->timer_entry.id = 0;

    /* Clear hash table entries */
    pj_hash_set(NULL, resolver->hquerybyid, &q->id, sizeof(q->id), 0, NULL);
    pj_hash_set(NULL, resolver->hquerybyres, &q->key, sizeof(q->key), 0, NULL);

    /* Notify applications first, to allow application to modify the 
     * record before it is saved to the hash table.
     */
    if (q->cb)
	(*q->cb)(q->user_data, status, dns_pkt);

    /* If query has subqueries, notify subqueries's application callback */
    if (!pj_list_empty(&q->child_head)) {
	pj_dns_async_query *child_q;

	child_q = q->child_head.next;
	while (child_q != (pj_dns_async_query*)&q->child_head) {
	    if (child_q->cb)
		(*child_q->cb)(child_q->user_data, status, dns_pkt);
	    child_q = child_q->next;
	}
    }

    /* Save/update response cache. */
    update_res_cache(resolver, &q->key, status, PJ_TRUE, dns_pkt);
    
    /* Recycle query objects, starting with the child queries */
    if (!pj_list_empty(&q->child_head)) {
	pj_dns_async_query *child_q;

	child_q = q->child_head.next;
	while (child_q != (pj_dns_async_query*)&q->child_head) {
	    pj_dns_async_query *next = child_q->next;
	    pj_list_erase(child_q);
	    pj_list_push_back(&resolver->query_free_nodes, child_q);
	    child_q = next;
	}
    }
    pj_list_push_back(&resolver->query_free_nodes, q);

read_next_packet:
    bytes_read = sizeof(resolver->udp_rx_pkt);
    resolver->udp_addr_len = sizeof(resolver->udp_src_addr);
    status = pj_ioqueue_recvfrom(resolver->udp_key, op_key, 
				 resolver->udp_rx_pkt,
				 &bytes_read, PJ_IOQUEUE_ALWAYS_ASYNC,
				 &resolver->udp_src_addr, 
				 &resolver->udp_addr_len);
    if (status != PJ_EPENDING) {
	char errmsg[PJ_ERR_MSG_SIZE];

	pj_strerror(status, errmsg, sizeof(errmsg));	
	PJ_LOG(4,(resolver->name.ptr, "DNS resolver ioqueue read error: %s",
		  errmsg));

	pj_assert(!"Unhandled error");
    }

    pj_mutex_unlock(resolver->mutex);
}


/*
 * Put the specified DNS packet into DNS cache. This function is mainly used
 * for testing the resolver, however it can also be used to inject entries
 * into the resolver.
 */
PJ_DEF(pj_status_t) pj_dns_resolver_add_entry( pj_dns_resolver *resolver,
					       const pj_dns_parsed_packet *pkt,
					       pj_bool_t set_ttl)
{
    struct res_key key;

    /* Sanity check */
    PJ_ASSERT_RETURN(resolver && pkt, PJ_EINVAL);

    /* Packet must be a DNS response */
    PJ_ASSERT_RETURN(PJ_DNS_GET_QR(pkt->hdr.flags) & 1, PJ_EINVAL);

    /* Make sure there are answers in the packet */
    PJ_ASSERT_RETURN((pkt->hdr.anscount && pkt->ans) ||
		      (pkt->hdr.qdcount && pkt->q),
		     PJLIB_UTIL_EDNSNOANSWERREC);

    pj_mutex_lock(resolver->mutex);

    /* Build resource key for looking up hash tables */
    pj_bzero(&key, sizeof(struct res_key));
    if (pkt->hdr.anscount) {
	/* Make sure name is not too long. */
	PJ_ASSERT_RETURN(pkt->ans[0].name.slen < PJ_MAX_HOSTNAME, 
			 PJ_ENAMETOOLONG);

	init_res_key(&key, pkt->ans[0].type, &pkt->ans[0].name);

    } else {
	/* Make sure name is not too long. */
	PJ_ASSERT_RETURN(pkt->q[0].name.slen < PJ_MAX_HOSTNAME, 
			 PJ_ENAMETOOLONG);

	init_res_key(&key, pkt->q[0].type, &pkt->q[0].name);
    }

    /* Insert entry. */
    update_res_cache(resolver, &key, PJ_SUCCESS, set_ttl, pkt);

    pj_mutex_unlock(resolver->mutex);

    return PJ_SUCCESS;
}


/*
 * Get the total number of response in the response cache.
 */
PJ_DEF(unsigned) pj_dns_resolver_get_cached_count(pj_dns_resolver *resolver)
{
    unsigned count;

    PJ_ASSERT_RETURN(resolver, 0);

    pj_mutex_lock(resolver->mutex);
    count = pj_hash_count(resolver->hrescache);
    pj_mutex_unlock(resolver->mutex);

    return count;
}


/*
 * Dump resolver state to the log.
 */
PJ_DEF(void) pj_dns_resolver_dump(pj_dns_resolver *resolver,
				  pj_bool_t detail)
{
#if PJ_LOG_MAX_LEVEL >= 3
    unsigned i;
    pj_time_val now;

    pj_mutex_lock(resolver->mutex);

    pj_gettimeofday(&now);

    PJ_LOG(3,(resolver->name.ptr, " Dumping resolver state:"));

    PJ_LOG(3,(resolver->name.ptr, "  Name servers:"));
    for (i=0; i<resolver->ns_count; ++i) {
	const char *state_names[] = { "probing", "active", "bad"};
	struct nameserver *ns = &resolver->ns[i];

	PJ_LOG(3,(resolver->name.ptr,
		  "   NS %d: %s:%d (state=%s until %ds, rtt=%d ms)",
		  i, pj_inet_ntoa(ns->addr.sin_addr),
		  pj_ntohs(ns->addr.sin_port),
		  state_names[ns->state],
		  ns->state_expiry.sec - now.sec,
		  PJ_TIME_VAL_MSEC(ns->rt_delay)));
    }

    PJ_LOG(3,(resolver->name.ptr, "  Nb. of cached responses: %u",
	      pj_hash_count(resolver->hrescache)));
    if (detail) {
	pj_hash_iterator_t itbuf, *it;
	it = pj_hash_first(resolver->hrescache, &itbuf);
	while (it) {
	    struct cached_res *cache = pj_hash_this(resolver->hrescache, it);
	    PJ_LOG(3,(resolver->name.ptr, 
		      "   Type %s: %s",
		      pj_dns_get_type_name(cache->key.qtype), 
		      cache->key.name));
	    it = pj_hash_next(resolver->hrescache, it);
	}
    }
    PJ_LOG(3,(resolver->name.ptr, "  Nb. of cached response free nodes: %u",
	      pj_list_size(&resolver->res_free_nodes)));
    PJ_LOG(3,(resolver->name.ptr, "  Nb. of pending queries: %u (%u)",
	      pj_hash_count(resolver->hquerybyid),
	      pj_hash_count(resolver->hquerybyres)));
    if (detail) {
	pj_hash_iterator_t itbuf, *it;
	it = pj_hash_first(resolver->hquerybyid, &itbuf);
	while (it) {
	    struct pj_dns_async_query *q;
	    q = pj_hash_this(resolver->hquerybyid, it);
	    PJ_LOG(3,(resolver->name.ptr, 
		      "   Type %s: %s",
		      pj_dns_get_type_name(q->key.qtype), 
		      q->key.name));
	    it = pj_hash_next(resolver->hquerybyid, it);
	}
    }
    PJ_LOG(3,(resolver->name.ptr, "  Nb. of pending query free nodes: %u",
	      pj_list_size(&resolver->query_free_nodes)));
    PJ_LOG(3,(resolver->name.ptr, "  Nb. of timer entries: %u",
	      pj_timer_heap_count(resolver->timer)));
    PJ_LOG(3,(resolver->name.ptr, "  Pool capacity: %d, used size: %d",
	      pj_pool_get_capacity(resolver->pool),
	      pj_pool_get_used_size(resolver->pool)));

    pj_mutex_unlock(resolver->mutex);
#endif
}