2 * Copyright (C) 1997-2008 Andre Noll <maan@systemlinux.org>
4 * Licensed under the GPL v2. For licencing details see COPYING.
7 /** \file vss.c The virtual streaming system.
9 * This contains the audio streaming code of para_server which is independent
10 * of the current audio format, audio file selector and of the activated
23 #include "server.cmdline.h"
31 extern struct misc_meta_data
*mmd
;
33 extern void dccp_send_init(struct sender
*);
34 extern void http_send_init(struct sender
*);
35 extern void ortp_send_init(struct sender
*);
37 /** The list of supported senders. */
38 struct sender senders
[] = {
41 .init
= http_send_init
,
45 .init
= dccp_send_init
,
50 .init
= ortp_send_init
,
58 /** The possible states of the afs socket. */
59 enum afs_socket_status
{
60 /** Socket is inactive. */
62 /** Socket fd was included in the write fd set for select(). */
63 AFS_SOCKET_CHECK_FOR_WRITE
,
64 /** vss wrote a request to the socket and waits for reply from afs. */
65 AFS_SOCKET_AFD_PENDING
68 /** The task structure for the virtual streaming system. */
70 struct timeval announce_tv
;
71 struct timeval data_send_barrier
;
72 struct timeval eof_barrier
;
73 struct timeval autoplay_barrier
;
75 enum afs_socket_status afsss
;
78 const char *header_buf
;
83 * Check if vss status flag \a P (playing) is set.
85 * \return Greater than zero if playing, zero otherwise.
88 unsigned int vss_playing(void)
90 return mmd
->new_vss_status_flags
& VSS_PLAYING
;
94 * Check if the \a N (next) status flag is set.
96 * \return Greater than zero if set, zero if not.
99 unsigned int vss_next(void)
101 return mmd
->new_vss_status_flags
& VSS_NEXT
;
105 * Check if a reposition request is pending.
107 * \return Greater than zero if true, zero otherwise.
110 unsigned int vss_repos(void)
112 return mmd
->new_vss_status_flags
& VSS_REPOS
;
116 * Check if the vss is currently paused.
118 * \return Greater than zero if paused, zero otherwise.
121 unsigned int vss_paused(void)
123 return !(mmd
->new_vss_status_flags
& VSS_NEXT
)
124 && !(mmd
->new_vss_status_flags
& VSS_PLAYING
);
128 * Check if the vss is currently stopped.
130 * \return Greater than zero if paused, zero otherwise.
133 unsigned int vss_stopped(void)
135 return (mmd
->new_vss_status_flags
& VSS_NEXT
)
136 && !(mmd
->new_vss_status_flags
& VSS_PLAYING
);
139 static int chk_barrier(const char *bname
, const struct timeval
*barrier
,
140 struct timeval
*diff
, int print_log
)
144 if (tv_diff(now
, barrier
, diff
) > 0)
148 PARA_DEBUG_LOG("%s barrier: %lims left\n", bname
, ms
);
153 * != NULL: timeout for next chunk
154 * NULL: nothing to do
156 static struct timeval
*vss_compute_timeout(struct vss_task
*vsst
)
158 static struct timeval the_timeout
;
159 struct timeval next_chunk
;
161 if (vss_next() && vsst
->map
) {
162 /* only sleep a bit, nec*/
163 the_timeout
.tv_sec
= 0;
164 the_timeout
.tv_usec
= 100;
167 if (chk_barrier("autoplay_delay", &vsst
->autoplay_barrier
,
168 &the_timeout
, 1) < 0)
170 if (chk_barrier("eof", &vsst
->eof_barrier
, &the_timeout
, 1) < 0)
172 if (chk_barrier("data send", &vsst
->data_send_barrier
,
173 &the_timeout
, 1) < 0)
175 if (!vss_playing() || !vsst
->map
)
177 compute_chunk_time(mmd
->chunks_sent
, &mmd
->afd
.afhi
.chunk_tv
,
178 &mmd
->stream_start
, &next_chunk
);
179 if (chk_barrier("chunk", &next_chunk
, &the_timeout
, 0) < 0)
181 /* chunk is due or bof */
182 the_timeout
.tv_sec
= 0;
183 the_timeout
.tv_usec
= 0;
187 static void vss_eof(struct vss_task
*vsst
)
189 mmd
->stream_start
= *now
;
192 if (mmd
->new_vss_status_flags
& VSS_NOMORE
)
193 mmd
->new_vss_status_flags
= VSS_NEXT
;
194 tv_add(&mmd
->afd
.afhi
.eof_tv
, now
, &vsst
->eof_barrier
);
195 para_munmap(vsst
->map
, mmd
->size
);
197 mmd
->chunks_sent
= 0;
199 mmd
->afd
.afhi
.seconds_total
= 0;
200 mmd
->afd
.afhi
.chunk_tv
.tv_sec
= 0;
201 mmd
->afd
.afhi
.chunk_tv
.tv_usec
= 0;
202 free(mmd
->afd
.afhi
.chunk_table
);
203 mmd
->afd
.afhi
.chunk_table
= NULL
;
204 free(mmd
->afd
.afhi
.info_string
);
205 mmd
->afd
.afhi
.info_string
= make_message("%s:\n%s:\n%s:\n", status_item_list
[SI_AUDIO_FILE_INFO
],
206 status_item_list
[SI_TAGINFO1
], status_item_list
[SI_TAGINFO2
]);
207 make_empty_status_items(mmd
->afd
.verbose_ls_output
);
214 * Get the list of all supported audio formats.
216 * \return Aa space separated list of all supported audio formats
217 * It is not allocated at runtime, i.e. there is no need to free
218 * the returned string in the caller.
220 const char *supported_audio_formats(void)
222 return SUPPORTED_AUDIO_FORMATS
;
226 * Get the chunk time of the current audio file.
228 * \return A pointer to a struct containing the chunk time, or NULL,
229 * if currently no audio file is selected.
231 struct timeval
*vss_chunk_time(void)
233 if (mmd
->afd
.afhi
.chunk_tv
.tv_sec
== 0 &&
234 mmd
->afd
.afhi
.chunk_tv
.tv_usec
== 0)
236 return &mmd
->afd
.afhi
.chunk_tv
;
239 static int need_to_request_new_audio_file(struct vss_task
*vsst
)
241 if (vsst
->map
) /* have audio file */
243 if (!vss_playing()) /* don't need one */
245 if (mmd
->new_vss_status_flags
& VSS_NOMORE
)
247 if (vsst
->afsss
== AFS_SOCKET_AFD_PENDING
) /* already requested one */
253 * Compute the timeout for para_server's main select-loop.
255 * This function gets called from para_server to determine the timeout value
256 * for its main select loop.
258 * \param s Pointer to the server scheduler.
259 * \param t Pointer to the vss task structure.
261 * Before the timeout is computed, the current vss status flags are evaluated
262 * and acted upon by calling appropriate functions from the lower layers.
263 * Possible actions include
265 * - request a new audio file from afs,
266 * - shutdown of all senders (stop/pause command),
267 * - reposition the stream (ff/jmp command).
269 static void vss_pre_select(struct sched
*s
, struct task
*t
)
272 struct timeval
*tv
, diff
;
273 struct vss_task
*vsst
= container_of(t
, struct vss_task
, task
);
275 if (!vsst
->map
|| vss_next() || vss_paused() || vss_repos())
276 for (i
= 0; senders
[i
].name
; i
++)
277 senders
[i
].shutdown_clients();
280 else if (vss_paused()) {
281 if (mmd
->chunks_sent
)
282 tv_add(&mmd
->afd
.afhi
.eof_tv
, now
, &vsst
->eof_barrier
);
283 mmd
->chunks_sent
= 0;
284 } else if (vss_repos()) {
285 tv_add(now
, &vsst
->announce_tv
, &vsst
->data_send_barrier
);
286 tv_add(&mmd
->afd
.afhi
.eof_tv
, now
, &vsst
->eof_barrier
);
287 mmd
->chunks_sent
= 0;
288 mmd
->current_chunk
= mmd
->repos_request
;
289 mmd
->new_vss_status_flags
&= ~VSS_REPOS
;
292 if (need_to_request_new_audio_file(vsst
)) {
293 PARA_DEBUG_LOG("ready and playing, but no audio file\n");
294 para_fd_set(vsst
->afs_socket
, &s
->wfds
, &s
->max_fileno
);
295 vsst
->afsss
= AFS_SOCKET_CHECK_FOR_WRITE
;
297 para_fd_set(vsst
->afs_socket
, &s
->rfds
, &s
->max_fileno
);
298 for (i
= 0; senders
[i
].name
; i
++) {
299 if (!senders
[i
].pre_select
)
301 senders
[i
].pre_select(&s
->max_fileno
, &s
->rfds
, &s
->wfds
);
303 tv
= vss_compute_timeout(vsst
);
304 if (tv
&& tv_diff(tv
, &s
->timeout
, &diff
) < 0)
308 static int recv_afs_msg(int afs_socket
, int *fd
, uint32_t *code
, uint32_t *data
)
310 char control
[255], buf
[8];
311 struct msghdr msg
= {.msg_iov
= NULL
};
312 struct cmsghdr
*cmsg
;
318 iov
.iov_len
= sizeof(buf
);
321 msg
.msg_control
= control
;
322 msg
.msg_controllen
= sizeof(control
);
323 memset(buf
, 0, sizeof(buf
));
324 ret
= recvmsg(afs_socket
, &msg
, 0);
326 return -ERRNO_TO_PARA_ERROR(errno
);
327 if (iov
.iov_len
!= sizeof(buf
))
328 return -E_AFS_SHORT_READ
;
329 *code
= *(uint32_t*)buf
;
330 *data
= *(uint32_t*)(buf
+ 4);
331 for (cmsg
= CMSG_FIRSTHDR(&msg
); cmsg
; cmsg
= CMSG_NXTHDR(&msg
, cmsg
)) {
332 if (cmsg
->cmsg_level
!= SOL_SOCKET
333 || cmsg
->cmsg_type
!= SCM_RIGHTS
)
335 if ((cmsg
->cmsg_len
- CMSG_LEN(0)) / sizeof(int) != 1)
337 *fd
= *(int *)CMSG_DATA(cmsg
);
342 static void recv_afs_result(struct vss_task
*vsst
)
344 int ret
, passed_fd
, shmid
;
345 uint32_t afs_code
= 0, afs_data
= 0;
348 vsst
->afsss
= AFS_SOCKET_READY
;
349 mmd
->afd
.afhi
.chunk_table
= NULL
;
350 ret
= recv_afs_msg(vsst
->afs_socket
, &passed_fd
, &afs_code
, &afs_data
);
353 PARA_DEBUG_LOG("fd: %d, code: %u, shmid: %u\n", passed_fd
, afs_code
,
356 if (afs_code
!= NEXT_AUDIO_FILE
)
361 ret
= load_afd(shmid
, &mmd
->afd
);
365 ret
= fstat(passed_fd
, &statbuf
);
367 PARA_ERROR_LOG("fstat error:\n");
368 ret
= -ERRNO_TO_PARA_ERROR(errno
);
371 mmd
->size
= statbuf
.st_size
;
372 mmd
->mtime
= statbuf
.st_mtime
;
373 vsst
->map
= para_mmap(mmd
->size
, PROT_READ
, MAP_PRIVATE
,
376 mmd
->chunks_sent
= 0;
377 mmd
->current_chunk
= 0;
381 mmd
->new_vss_status_flags
&= (~VSS_NEXT
);
382 tv_add(now
, &vsst
->announce_tv
, &vsst
->data_send_barrier
);
383 afh_get_header(&mmd
->afd
.afhi
, vsst
->map
, &vsst
->header_buf
,
387 free(mmd
->afd
.afhi
.chunk_table
);
390 PARA_ERROR_LOG("%s\n", para_strerror(-ret
));
391 mmd
->new_vss_status_flags
= VSS_NEXT
;
395 * Main sending function.
397 * This function gets called from para_server as soon as the next chunk of data
398 * should be pushed out. It obtains a pointer to the data to be sent out as
399 * well as its length from mmd->afd.afhi. This information is then passed to
400 * each supported sender's send() function which is supposed to send out the data
401 * to all connected clients.
403 static void vss_send_chunk(struct vss_task
*vsst
)
410 if (!vsst
->map
|| !vss_playing())
412 compute_chunk_time(mmd
->chunks_sent
, &mmd
->afd
.afhi
.chunk_tv
,
413 &mmd
->stream_start
, &due
);
414 if (tv_diff(&due
, now
, NULL
) > 0)
416 if (chk_barrier("eof", &vsst
->eof_barrier
, &due
, 1) < 0)
418 if (chk_barrier("data send", &vsst
->data_send_barrier
,
421 if (mmd
->current_chunk
>= mmd
->afd
.afhi
.chunks_total
) { /* eof */
422 mmd
->new_vss_status_flags
|= VSS_NEXT
;
426 * We call the send function also in case of empty chunks as they
427 * might have still some data queued which can be sent in this case.
429 if (!mmd
->chunks_sent
) {
431 mmd
->stream_start
= *now
;
432 tv_scale(mmd
->current_chunk
, &mmd
->afd
.afhi
.chunk_tv
, &tmp
);
433 mmd
->offset
= tv2ms(&tmp
);
436 afh_get_chunk(mmd
->current_chunk
, &mmd
->afd
.afhi
, vsst
->map
, &buf
, &len
);
437 for (i
= 0; senders
[i
].name
; i
++)
438 senders
[i
].send(mmd
->current_chunk
, mmd
->chunks_sent
, buf
, len
,
439 vsst
->header_buf
, vsst
->header_len
);
440 mmd
->new_vss_status_flags
|= VSS_PLAYING
;
442 mmd
->current_chunk
++;
445 static void vss_post_select(struct sched
*s
, struct task
*t
)
448 struct vss_task
*vsst
= container_of(t
, struct vss_task
, task
);
450 if (mmd
->sender_cmd_data
.cmd_num
>= 0) {
451 int num
= mmd
->sender_cmd_data
.cmd_num
,
452 sender_num
= mmd
->sender_cmd_data
.sender_num
;
454 if (senders
[sender_num
].client_cmds
[num
])
455 senders
[sender_num
].client_cmds
[num
](&mmd
->sender_cmd_data
);
456 mmd
->sender_cmd_data
.cmd_num
= -1;
458 if (vsst
->afsss
!= AFS_SOCKET_CHECK_FOR_WRITE
) {
459 if (FD_ISSET(vsst
->afs_socket
, &s
->rfds
))
460 recv_afs_result(vsst
);
461 } else if (FD_ISSET(vsst
->afs_socket
, &s
->wfds
)) {
462 PARA_NOTICE_LOG("requesting new fd from afs\n");
463 ret
= send_buffer(vsst
->afs_socket
, "new");
464 vsst
->afsss
= AFS_SOCKET_AFD_PENDING
;
466 for (i
= 0; senders
[i
].name
; i
++) {
467 if (!senders
[i
].post_select
)
469 senders
[i
].post_select(&s
->rfds
, &s
->wfds
);
471 vss_send_chunk(vsst
);
475 * Initialize the virtual streaming system task.
477 * \param afs_socket The fd for communication with afs.
479 * This also initializes all supported senders and starts streaming
480 * if the --autoplay command line flag was given.
482 void init_vss_task(int afs_socket
)
484 static struct vss_task vss_task_struct
, *vsst
= &vss_task_struct
;
486 char *hn
= para_hostname(), *home
= para_homedir();
487 long unsigned announce_time
= conf
.announce_time_arg
> 0?
488 conf
.announce_time_arg
: 300,
489 autoplay_delay
= conf
.autoplay_delay_arg
> 0?
490 conf
.autoplay_delay_arg
: 0;
492 vsst
->afs_socket
= afs_socket
;
493 vsst
->task
.pre_select
= vss_pre_select
;
494 vsst
->task
.post_select
= vss_post_select
;
495 ms2tv(announce_time
, &vsst
->announce_tv
);
496 PARA_INFO_LOG("announce timeval: %lums\n", tv2ms(&vsst
->announce_tv
));
497 for (i
= 0; senders
[i
].name
; i
++) {
498 PARA_NOTICE_LOG("initializing %s sender\n", senders
[i
].name
);
499 senders
[i
].init(&senders
[i
]);
503 mmd
->sender_cmd_data
.cmd_num
= -1;
504 if (conf
.autoplay_given
) {
506 mmd
->vss_status_flags
|= VSS_PLAYING
;
507 mmd
->new_vss_status_flags
|= VSS_PLAYING
;
508 ms2tv(autoplay_delay
, &tmp
);
509 tv_add(now
, &tmp
, &vsst
->autoplay_barrier
);
511 register_task(&vsst
->task
);