1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
/*
* Copyright (C) 2007-2009 Andre Noll <maan@systemlinux.org>
*
* Licensed under the GPL v2. For licencing details see COPYING.
*/
/** \file chunk_queue.c Queuing functions for paraslash senders. */
#include "para.h"
#include "list.h"
#include "afh.h"
#include "vss.h"
#include "string.h"
#include "error.h"
/**
* Senders may use the chunk queue facility to deal with laggy connections. It
* allows them to enqueue chunks if they can not be sent out immediately.
*
* Chunk queues are "cheap" in the sense that only reference to the audio file
* data is stored, but not the data itself.
*/
struct chunk_queue {
/** The list of pending chunks for this client. */
struct list_head q;
/** The number of pending bytes for this client. */
unsigned long num_pending;
/** More than that many bytes in the queue is considered an error. */
unsigned long max_pending;
};
/** Describes one queued chunk in a chunk queue. */
struct queued_chunk {
/** Pointer to the data to be queued. */
const char *buf;
/** The number of bytes of this chunk. */
size_t num_bytes;
/** Position of the chunk in the chunk queue. */
struct list_head node;
};
/**
* Add a chunk to the given queue.
*
* \param cq the queue to add the chunk to.
* \param buf Pointer to the data to be queued.
* \param num_bytes The size of \a buf.
*
* \return Standard.
*/
int cq_enqueue(struct chunk_queue *cq, const char *buf, size_t num_bytes)
{
struct queued_chunk *qc;
if (cq->num_pending + num_bytes > cq->max_pending)
return -E_QUEUE;
qc = para_malloc(sizeof(struct queued_chunk));
cq->num_pending += num_bytes;
qc->buf = buf;
qc->num_bytes = num_bytes;
list_add_tail(&qc->node, &cq->q);
PARA_DEBUG_LOG("%lu bytes queued for %p\n", cq->num_pending, &cq->q);
return 1;
}
/**
* Lookup the next chunk in the queue.
*
* \param cq The chunk queue.
*
* \return The next queued chunk, or \p NULL if there is no chunk available.
*/
struct queued_chunk *cq_peek(struct chunk_queue *cq)
{
if (list_empty(&cq->q))
return NULL;
return list_entry(cq->q.next, struct queued_chunk, node);
}
/**
* Remove the current chunk from the queue.
*
* \param cq The queue to remove from.
*/
void cq_dequeue(struct chunk_queue *cq)
{
struct queued_chunk *qc = cq_peek(cq);
assert(qc);
assert(cq->num_pending >= qc->num_bytes);
cq->num_pending -= qc->num_bytes;
list_del(&qc->node);
free(qc);
}
/**
* Change the number of bytes sent for the current queued chunk.
*
* \param cq The chunk queue.
* \param sent Number of bytes successfully sent.
*/
void cq_update(struct chunk_queue *cq, size_t sent)
{
struct queued_chunk *qc = cq_peek(cq);
assert(qc);
qc->num_bytes -= sent;
qc->buf += sent;
cq->num_pending -= sent;
}
/**
* Get a pointer to the given queued chunk.
*
* \param qc The queued chunk.
* \param buf Result pointer.
* \param num_bytes Number of bytes of \a buf.
*
* \return Positive on success, negative on errors.
*/
int cq_get(struct queued_chunk *qc, const char **buf, size_t *num_bytes)
{
*buf = qc->buf;
*num_bytes = qc->num_bytes;
return 1;
}
/**
* Allocate and initialize a chunk queue.
*
* \param max_pending Maximal number of bytes that will be queued.
*
* \return A pointer to the new queue.
*/
struct chunk_queue *cq_new(size_t max_pending)
{
struct chunk_queue *cq = para_malloc(sizeof(*cq));
INIT_LIST_HEAD(&cq->q);
cq->max_pending = max_pending;
cq->num_pending = 0;
return cq;
}
/**
* Deallocate all resources of this queue.
*
* \param cq The chunk queue.
*/
void cq_destroy(struct chunk_queue *cq)
{
struct queued_chunk *qc, *tmp;
list_for_each_entry_safe(qc, tmp, &cq->q, node) {
list_del(&qc->node);
free(qc);
}
free(cq);
}
|
