+ats_reserve_callback (void *cls,
+ const struct GNUNET_PeerIdentity *peer,
+ int32_t amount,
+ struct GNUNET_TIME_Relative res_delay);
+
+
+/**
+ * If ready (bandwidth reserved), try to schedule transmission via
+ * core for the given handle.
+ *
+ * @param pth transmission handle to schedule
+ */
+static void
+schedule_transmission (struct GSF_PeerTransmitHandle *pth)
+{
+ struct GSF_ConnectedPeer *cp;
+ struct GNUNET_PeerIdentity target;
+
+ cp = pth->cp;
+ if ((NULL != cp->cth) || (0 != cp->cth_in_progress))
+ return; /* already done */
+ GNUNET_assert (0 != cp->ppd.pid);
+ GNUNET_PEER_resolve (cp->ppd.pid, &target);
+
+ if (0 != cp->inc_preference)
+ {
+ GNUNET_ATS_performance_change_preference (GSF_ats,
+ &target,
+ GNUNET_ATS_PREFERENCE_BANDWIDTH,
+ (double) cp->inc_preference,
+ GNUNET_ATS_PREFERENCE_END);
+ cp->inc_preference = 0;
+ }
+
+ if ( (GNUNET_YES == pth->is_query) &&
+ (GNUNET_YES != pth->was_reserved) )
+ {
+ /* query, need reservation */
+ if (GNUNET_YES != cp->did_reserve)
+ return; /* not ready */
+ cp->did_reserve = GNUNET_NO;
+ /* reservation already done! */
+ pth->was_reserved = GNUNET_YES;
+ cp->rc = GNUNET_ATS_reserve_bandwidth (GSF_ats,
+ &target,
+ DBLOCK_SIZE,
+ &ats_reserve_callback,
+ cp);
+ return;
+ }
+ GNUNET_assert (NULL == cp->cth);
+ cp->cth_in_progress++;
+ cp->cth =
+ GNUNET_CORE_notify_transmit_ready (GSF_core,
+ GNUNET_YES,
+ GNUNET_CORE_PRIO_BACKGROUND,
+ GNUNET_TIME_absolute_get_remaining (pth->timeout),
+ &target,
+ pth->size,
+ &peer_transmit_ready_cb, cp);
+ GNUNET_assert (NULL != cp->cth);
+ GNUNET_assert (0 < cp->cth_in_progress--);
+}
+
+
+/**
+ * Core is ready to transmit to a peer, get the message.
+ *
+ * @param cls the `struct GSF_PeerTransmitHandle` of the message
+ * @param size number of bytes core is willing to take
+ * @param buf where to copy the message
+ * @return number of bytes copied to @a buf
+ */
+static size_t
+peer_transmit_ready_cb (void *cls,
+ size_t size,
+ void *buf)
+{
+ struct GSF_ConnectedPeer *cp = cls;
+ struct GSF_PeerTransmitHandle *pth = cp->pth_head;
+ struct GSF_PeerTransmitHandle *pos;
+ size_t ret;
+
+ cp->cth = NULL;
+ if (NULL == pth)
+ return 0;
+ if (pth->size > size)
+ {
+ schedule_transmission (pth);
+ return 0;
+ }
+ if (NULL != pth->timeout_task)
+ {
+ GNUNET_SCHEDULER_cancel (pth->timeout_task);
+ pth->timeout_task = NULL;
+ }
+ GNUNET_CONTAINER_DLL_remove (cp->pth_head,
+ cp->pth_tail,
+ pth);
+ if (GNUNET_YES == pth->is_query)
+ {
+ cp->ppd.last_request_times[(cp->last_request_times_off++) %
+ MAX_QUEUE_PER_PEER] =
+ GNUNET_TIME_absolute_get ();
+ GNUNET_assert (0 < cp->ppd.pending_queries--);
+ }
+ else if (GNUNET_NO == pth->is_query)
+ {
+ GNUNET_assert (0 < cp->ppd.pending_replies--);
+ }
+ GNUNET_LOAD_update (cp->ppd.transmission_delay,
+ GNUNET_TIME_absolute_get_duration
+ (pth->transmission_request_start_time).rel_value_us);
+ ret = pth->gmc (pth->gmc_cls, size, buf);
+ if (NULL != (pos = cp->pth_head))
+ {
+ GNUNET_assert (pos != pth);
+ schedule_transmission (pos);
+ }
+ GNUNET_free (pth);
+ return ret;
+}