FFmpeg: libavformat/udp.c Source File

114 {"udplite_coverage", "choose UDPLite head size which should be validated by checksum", OFFSET(udplite_coverage), AV_OPT_TYPE_INT, {.i64 = 0}, 0, INT_MAX, D|E },

115 {"pkt_size", "set size of UDP packets", OFFSET(packet_size), AV_OPT_TYPE_INT, {.i64 = 1472}, 0, INT_MAX, D|E },

116 {"reuse", "explicitly allow or disallow reusing UDP sockets", OFFSET(reuse_socket), AV_OPT_TYPE_INT, {.i64 = 0}, 0, 1, D|E },

117 {"broadcast", "explicitly allow or disallow broadcast destination", OFFSET(is_broadcast), AV_OPT_TYPE_INT, {.i64 = 0}, 0, 1, E },

118 {"ttl", "set the time to live value (for multicast only)", OFFSET(ttl), AV_OPT_TYPE_INT, {.i64 = 16}, 0, INT_MAX, E },

119 {"connect", "set if connect() should be called on socket", OFFSET(is_connected), AV_OPT_TYPE_INT, {.i64 = 0}, 0, 1, D|E },

120 /* TODO 'sources', 'block' option */

121 {"fifo_size", "set the UDP receiving circular buffer size, expressed as a number of packets with size of 188 bytes", OFFSET(circular_buffer_size), AV_OPT_TYPE_INT, {.i64 = 7*4096}, 0, INT_MAX, D },

122 {"overrun_nonfatal", "survive in case of UDP receiving circular buffer overrun", OFFSET(overrun_nonfatal), AV_OPT_TYPE_INT, {.i64 = 0}, 0, 1, D },

123 {"timeout", "set raise error timeout (only in read mode)", OFFSET(timeout), AV_OPT_TYPE_INT, {.i64 = 0}, 0, INT_MAX, D },

124 {NULL}

125 };

126

127 static const AVClass udp_context_class = {

128 .class_name = "udp",

129 .item_name = av_default_item_name,

130 .option = options,

131 .version = LIBAVUTIL_VERSION_INT,

132 };

133

134 static const AVClass udplite_context_class = {

135 .class_name = "udplite",

136 .item_name = av_default_item_name,

137 .option = options,

138 .version = LIBAVUTIL_VERSION_INT,

139 };

140

141 static void log_net_error(void *ctx, int level, const char* prefix)

142 {

143 char errbuf[100];

144 av_strerror(ff_neterrno(), errbuf, sizeof(errbuf));

145 av_log(ctx, level, "%s: %s\n", prefix, errbuf);

146 }

147

148 static int udp_set_multicast_ttl(int sockfd, int mcastTTL,

149 struct sockaddr *addr)

150 {

151 #ifdef IP_MULTICAST_TTL

152 if (addr->sa_family == AF_INET) {

153 if (setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_TTL, &mcastTTL, sizeof(mcastTTL)) < 0) {

154 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_MULTICAST_TTL)");

155 return -1;

156 }

157 }

158 #endif

159 #if defined(IPPROTO_IPV6) && defined(IPV6_MULTICAST_HOPS)

160 if (addr->sa_family == AF_INET6) {

161 if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &mcastTTL, sizeof(mcastTTL)) < 0) {

162 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IPV6_MULTICAST_HOPS)");

163 return -1;

164 }

165 }

166 #endif

167 return 0;

168 }

169

170 static int udp_join_multicast_group(int sockfd, struct sockaddr *addr,struct sockaddr *local_addr)

171 {

172 #ifdef IP_ADD_MEMBERSHIP

173 if (addr->sa_family == AF_INET) {

174 struct ip_mreq mreq;

175

176 mreq.imr_multiaddr.s_addr = ((struct sockaddr_in *)addr)->sin_addr.s_addr;

177 if (local_addr)

178 mreq.imr_interface= ((struct sockaddr_in *)local_addr)->sin_addr;

179 else

180 mreq.imr_interface.s_addr= INADDR_ANY;

181 if (setsockopt(sockfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, (const void *)&mreq, sizeof(mreq)) < 0) {

182 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_ADD_MEMBERSHIP)");

183 return -1;

184 }

185 }

186 #endif

187 #if HAVE_STRUCT_IPV6_MREQ && defined(IPPROTO_IPV6)

188 if (addr->sa_family == AF_INET6) {

189 struct ipv6_mreq mreq6;

190

191 memcpy(&mreq6.ipv6mr_multiaddr, &(((struct sockaddr_in6 *)addr)->sin6_addr), sizeof(struct in6_addr));

192 mreq6.ipv6mr_interface= 0;

193 if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, &mreq6, sizeof(mreq6)) < 0) {

194 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IPV6_ADD_MEMBERSHIP)");

195 return -1;

196 }

197 }

198 #endif

199 return 0;

200 }

201

202 static int udp_leave_multicast_group(int sockfd, struct sockaddr *addr,struct sockaddr *local_addr)

203 {

204 #ifdef IP_DROP_MEMBERSHIP

205 if (addr->sa_family == AF_INET) {

206 struct ip_mreq mreq;

207

208 mreq.imr_multiaddr.s_addr = ((struct sockaddr_in *)addr)->sin_addr.s_addr;

209 if (local_addr)

210 mreq.imr_interface= ((struct sockaddr_in *)local_addr)->sin_addr;

211 else

212 mreq.imr_interface.s_addr= INADDR_ANY;

213 if (setsockopt(sockfd, IPPROTO_IP, IP_DROP_MEMBERSHIP, (const void *)&mreq, sizeof(mreq)) < 0) {

214 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_DROP_MEMBERSHIP)");

215 return -1;

216 }

217 }

218 #endif

219 #if HAVE_STRUCT_IPV6_MREQ && defined(IPPROTO_IPV6)

220 if (addr->sa_family == AF_INET6) {

221 struct ipv6_mreq mreq6;

222

223 memcpy(&mreq6.ipv6mr_multiaddr, &(((struct sockaddr_in6 *)addr)->sin6_addr), sizeof(struct in6_addr));

224 mreq6.ipv6mr_interface= 0;

225 if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_DROP_MEMBERSHIP, &mreq6, sizeof(mreq6)) < 0) {

226 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IPV6_DROP_MEMBERSHIP)");

227 return -1;

228 }

229 }

230 #endif

231 return 0;

232 }

233

234 static struct addrinfo* udp_resolve_host(const char *hostname, int port,

235 int type, int family, int flags)

236 {

237 struct addrinfo hints = { 0 }, *res = 0;

238 int error;

239 char sport[16];

240 const char *node = 0, *service = "0";

241

242 if (port > 0) {

243 snprintf(sport, sizeof(sport), "%d", port);

244 service = sport;

245 }

246 if ((hostname) && (hostname[0] != '0円') && (hostname[0] != '?')) {

247 node = hostname;

248 }

249 hints.ai_socktype = type;

250 hints.ai_family = family;

251 hints.ai_flags = flags;

252 if ((error = getaddrinfo(node, service, &hints, &res))) {

253 res = NULL;

254 av_log(NULL, AV_LOG_ERROR, "udp_resolve_host: %s\n", gai_strerror(error));

255 }

256

257 return res;

258 }

259

260 static int udp_set_multicast_sources(int sockfd, struct sockaddr *addr,

261 int addr_len, char **sources,

262 int nb_sources, int include)

263 {

264 #if HAVE_STRUCT_GROUP_SOURCE_REQ && defined(MCAST_BLOCK_SOURCE) && !defined(_WIN32)

265 /* These ones are available in the microsoft SDK, but don't seem to work

266 * as on linux, so just prefer the v4-only approach there for now. */

267 int i;

268 for (i = 0; i < nb_sources; i++) {

269 struct group_source_req mreqs;

270 int level = addr->sa_family == AF_INET ? IPPROTO_IP : IPPROTO_IPV6;

271 struct addrinfo *sourceaddr = udp_resolve_host(sources[i], 0,

272 SOCK_DGRAM, AF_UNSPEC,

273 0);

274 if (!sourceaddr)

275 return AVERROR(ENOENT);

276

277 mreqs.gsr_interface = 0;

278 memcpy(&mreqs.gsr_group, addr, addr_len);

279 memcpy(&mreqs.gsr_source, sourceaddr->ai_addr, sourceaddr->ai_addrlen);

280 freeaddrinfo(sourceaddr);

281

282 if (setsockopt(sockfd, level,

283 include ? MCAST_JOIN_SOURCE_GROUP : MCAST_BLOCK_SOURCE,

284 (const void *)&mreqs, sizeof(mreqs)) < 0) {

285 if (include)

286 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(MCAST_JOIN_SOURCE_GROUP)");

287 else

288 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(MCAST_BLOCK_SOURCE)");

289 return ff_neterrno();

290 }

291 }

292 #elif HAVE_STRUCT_IP_MREQ_SOURCE && defined(IP_BLOCK_SOURCE)

293 int i;

294 if (addr->sa_family != AF_INET) {

295 av_log(NULL, AV_LOG_ERROR,

296 "Setting multicast sources only supported for IPv4\n");

297 return AVERROR(EINVAL);

298 }

299 for (i = 0; i < nb_sources; i++) {

300 struct ip_mreq_source mreqs;

301 struct addrinfo *sourceaddr = udp_resolve_host(sources[i], 0,

302 SOCK_DGRAM, AF_UNSPEC,

303 0);

304 if (!sourceaddr)

305 return AVERROR(ENOENT);

306 if (sourceaddr->ai_addr->sa_family != AF_INET) {

307 freeaddrinfo(sourceaddr);

308 av_log(NULL, AV_LOG_ERROR, "%s is of incorrect protocol family\n",

309 sources[i]);

310 return AVERROR(EINVAL);

311 }

312

313 mreqs.imr_multiaddr.s_addr = ((struct sockaddr_in *)addr)->sin_addr.s_addr;

314 mreqs.imr_interface.s_addr = INADDR_ANY;

315 mreqs.imr_sourceaddr.s_addr = ((struct sockaddr_in *)sourceaddr->ai_addr)->sin_addr.s_addr;

316 freeaddrinfo(sourceaddr);

317

318 if (setsockopt(sockfd, IPPROTO_IP,

319 include ? IP_ADD_SOURCE_MEMBERSHIP : IP_BLOCK_SOURCE,

320 (const void *)&mreqs, sizeof(mreqs)) < 0) {

321 if (include)

322 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_ADD_SOURCE_MEMBERSHIP)");

323 else

324 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_BLOCK_SOURCE)");

325 return ff_neterrno();

326 }

327 }

328 #else

329 return AVERROR(ENOSYS);

330 #endif

331 return 0;

332 }

333 static int udp_set_url(struct sockaddr_storage *addr,

334 const char *hostname, int port)

335 {

336 struct addrinfo *res0;

337 int addr_len;

338

339 res0 = udp_resolve_host(hostname, port, SOCK_DGRAM, AF_UNSPEC, 0);

340 if (!res0) return AVERROR(EIO);

341 memcpy(addr, res0->ai_addr, res0->ai_addrlen);

342 addr_len = res0->ai_addrlen;

343 freeaddrinfo(res0);

344

345 return addr_len;

346 }

347

348 static int udp_socket_create(UDPContext *s, struct sockaddr_storage *addr,

349 socklen_t *addr_len, const char *localaddr)

350 {

351 int udp_fd = -1;

352 struct addrinfo *res0, *res;

353 int family = AF_UNSPEC;

354

355 if (((struct sockaddr *) &s->dest_addr)->sa_family)

356 family = ((struct sockaddr *) &s->dest_addr)->sa_family;

357 res0 = udp_resolve_host(localaddr[0] ? localaddr : NULL, s->local_port,

358 SOCK_DGRAM, family, AI_PASSIVE);

359 if (!res0)

360 goto fail;

361 for (res = res0; res; res=res->ai_next) {

362 if (s->udplite_coverage)

363 udp_fd = ff_socket(res->ai_family, SOCK_DGRAM, IPPROTO_UDPLITE);

364 else

365 udp_fd = ff_socket(res->ai_family, SOCK_DGRAM, 0);

366 if (udp_fd != -1) break;

367 log_net_error(NULL, AV_LOG_ERROR, "socket");

368 }

369

370 if (udp_fd < 0)

371 goto fail;

372

373 memcpy(addr, res->ai_addr, res->ai_addrlen);

374 *addr_len = res->ai_addrlen;

375

376 freeaddrinfo(res0);

377

378 return udp_fd;

379

380 fail:

381 if (udp_fd >= 0)

382 closesocket(udp_fd);

383 if(res0)

384 freeaddrinfo(res0);

385 return -1;

386 }

387

388 static int udp_port(struct sockaddr_storage *addr, int addr_len)

389 {

390 char sbuf[sizeof(int)*3+1];

391 int error;

392

393 if ((error = getnameinfo((struct sockaddr *)addr, addr_len, NULL, 0, sbuf, sizeof(sbuf), NI_NUMERICSERV)) != 0) {

394 av_log(NULL, AV_LOG_ERROR, "getnameinfo: %s\n", gai_strerror(error));

395 return -1;

396 }

397

398 return strtol(sbuf, NULL, 10);

399 }

400

401

402 /**

403 * If no filename is given to av_open_input_file because you want to

404 * get the local port first, then you must call this function to set

405 * the remote server address.

406 *

407 * url syntax: udp://host:port[?option=val...]

408 * option: 'ttl=n' : set the ttl value (for multicast only)

409 * 'localport=n' : set the local port

410 * 'pkt_size=n' : set max packet size

411 * 'reuse=1' : enable reusing the socket

412 * 'overrun_nonfatal=1': survive in case of circular buffer overrun

413 *

414 * @param h media file context

415 * @param uri of the remote server

416 * @return zero if no error.

417 */

418 int ff_udp_set_remote_url(URLContext *h, const char *uri)

419 {

420 UDPContext *s = h->priv_data;

421 char hostname[256], buf[10];

422 int port;

423 const char *p;

424

425 av_url_split(NULL, 0, NULL, 0, hostname, sizeof(hostname), &port, NULL, 0, uri);

426

427 /* set the destination address */

428 s->dest_addr_len = udp_set_url(&s->dest_addr, hostname, port);

429 if (s->dest_addr_len < 0) {

430 return AVERROR(EIO);

431 }

432 s->is_multicast = ff_is_multicast_address((struct sockaddr*) &s->dest_addr);

433 p = strchr(uri, '?');

434 if (p) {

435 if (av_find_info_tag(buf, sizeof(buf), "connect", p)) {

436 int was_connected = s->is_connected;

437 s->is_connected = strtol(buf, NULL, 10);

438 if (s->is_connected && !was_connected) {

439 if (connect(s->udp_fd, (struct sockaddr *) &s->dest_addr,

440 s->dest_addr_len)) {

441 s->is_connected = 0;

442 log_net_error(h, AV_LOG_ERROR, "connect");

443 return AVERROR(EIO);

444 }

445 }

446 }

447 }

448

449 return 0;

450 }

451

452 /**

453 * Return the local port used by the UDP connection

454 * @param h media file context

455 * @return the local port number

456 */

457 int ff_udp_get_local_port(URLContext *h)

458 {

459 UDPContext *s = h->priv_data;

460 return s->local_port;

461 }

462

463 /**

464 * Return the udp file handle for select() usage to wait for several RTP

465 * streams at the same time.

466 * @param h media file context

467 */

468 static int udp_get_file_handle(URLContext *h)

469 {

470 UDPContext *s = h->priv_data;

471 return s->udp_fd;

472 }

473

474 #if HAVE_PTHREAD_CANCEL

475 static void *circular_buffer_task( void *_URLContext)

476 {

477 URLContext *h = _URLContext;

478 UDPContext *s = h->priv_data;

479 int old_cancelstate;

480

481 pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);

482 pthread_mutex_lock(&s->mutex);

483 if (ff_socket_nonblock(s->udp_fd, 0) < 0) {

484 av_log(h, AV_LOG_ERROR, "Failed to set blocking mode");

485 s->circular_buffer_error = AVERROR(EIO);

486 goto end;

487 }

488 while(1) {

489 int len;

490

491 pthread_mutex_unlock(&s->mutex);

492 /* Blocking operations are always cancellation points;

493 see "General Information" / "Thread Cancelation Overview"

494 in Single Unix. */

495 pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &old_cancelstate);

496 len = recv(s->udp_fd, s->tmp+4, sizeof(s->tmp)-4, 0);

497 pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);

498 pthread_mutex_lock(&s->mutex);

499 if (len < 0) {

500 if (ff_neterrno() != AVERROR(EAGAIN) && ff_neterrno() != AVERROR(EINTR)) {

501 s->circular_buffer_error = ff_neterrno();

502 goto end;

503 }

504 continue;

505 }

506 AV_WL32(s->tmp, len);

507

508 if(av_fifo_space(s->fifo) < len + 4) {

509 /* No Space left */

510 if (s->overrun_nonfatal) {

511 av_log(h, AV_LOG_WARNING, "Circular buffer overrun. "

512 "Surviving due to overrun_nonfatal option\n");

513 continue;

514 } else {

515 av_log(h, AV_LOG_ERROR, "Circular buffer overrun. "

516 "To avoid, increase fifo_size URL option. "

517 "To survive in such case, use overrun_nonfatal option\n");

518 s->circular_buffer_error = AVERROR(EIO);

519 goto end;

520 }

521 }

522 av_fifo_generic_write(s->fifo, s->tmp, len+4, NULL);

523 pthread_cond_signal(&s->cond);

524 }

525

526 end:

527 pthread_cond_signal(&s->cond);

528 pthread_mutex_unlock(&s->mutex);

529 return NULL;

530 }

531 #endif

532

533 static int parse_source_list(char *buf, char **sources, int *num_sources,

534 int max_sources)

535 {

536 char *source_start;

537

538 source_start = buf;

539 while (1) {

540 char *next = strchr(source_start, ',');

541 if (next)

542 *next = '0円';

543 sources[*num_sources] = av_strdup(source_start);

544 if (!sources[*num_sources])

545 return AVERROR(ENOMEM);

546 source_start = next + 1;

547 (*num_sources)++;

548 if (*num_sources >= max_sources || !next)

549 break;

550 }

551 return 0;

552 }

553

554 /* put it in UDP context */

555 /* return non zero if error */

556 static int udp_open(URLContext *h, const char *uri, int flags)

557 {

558 char hostname[1024], localaddr[1024] = "";

559 int port, udp_fd = -1, tmp, bind_ret = -1, dscp = -1;

560 UDPContext *s = h->priv_data;

561 int is_output;

562 const char *p;

563 char buf[256];

564 struct sockaddr_storage my_addr;

565 socklen_t len;

566 int reuse_specified = 0;

567 int i, num_include_sources = 0, num_exclude_sources = 0;

568 char *include_sources[32], *exclude_sources[32];

569

570 h->is_streamed = 1;

571

572 is_output = !(flags & AVIO_FLAG_READ);

573 if (!s->buffer_size) /* if not set explicitly */

574 s->buffer_size = is_output ? UDP_TX_BUF_SIZE : UDP_MAX_PKT_SIZE;

575

576 p = strchr(uri, '?');

577 if (p) {

578 if (av_find_info_tag(buf, sizeof(buf), "reuse", p)) {

579 char *endptr = NULL;

580 s->reuse_socket = strtol(buf, &endptr, 10);

581 /* assume if no digits were found it is a request to enable it */

582 if (buf == endptr)

583 s->reuse_socket = 1;

584 reuse_specified = 1;

585 }

586 if (av_find_info_tag(buf, sizeof(buf), "overrun_nonfatal", p)) {

587 char *endptr = NULL;

588 s->overrun_nonfatal = strtol(buf, &endptr, 10);

589 /* assume if no digits were found it is a request to enable it */

590 if (buf == endptr)

591 s->overrun_nonfatal = 1;

592 if (!HAVE_PTHREAD_CANCEL)

593 av_log(h, AV_LOG_WARNING,

594 "'overrun_nonfatal' option was set but it is not supported "

595 "on this build (pthread support is required)\n");

596 }

597 if (av_find_info_tag(buf, sizeof(buf), "ttl", p)) {

598 s->ttl = strtol(buf, NULL, 10);

599 }

600 if (av_find_info_tag(buf, sizeof(buf), "udplite_coverage", p)) {

601 s->udplite_coverage = strtol(buf, NULL, 10);

602 }

603 if (av_find_info_tag(buf, sizeof(buf), "localport", p)) {

604 s->local_port = strtol(buf, NULL, 10);

605 }

606 if (av_find_info_tag(buf, sizeof(buf), "pkt_size", p)) {

607 s->packet_size = strtol(buf, NULL, 10);

608 }

609 if (av_find_info_tag(buf, sizeof(buf), "buffer_size", p)) {

610 s->buffer_size = strtol(buf, NULL, 10);

611 }

612 if (av_find_info_tag(buf, sizeof(buf), "connect", p)) {

613 s->is_connected = strtol(buf, NULL, 10);

614 }

615 if (av_find_info_tag(buf, sizeof(buf), "dscp", p)) {

616 dscp = strtol(buf, NULL, 10);

617 }

618 if (av_find_info_tag(buf, sizeof(buf), "fifo_size", p)) {

619 s->circular_buffer_size = strtol(buf, NULL, 10);

620 if (!HAVE_PTHREAD_CANCEL)

621 av_log(h, AV_LOG_WARNING,

622 "'circular_buffer_size' option was set but it is not supported "

623 "on this build (pthread support is required)\n");

624 }

625 if (av_find_info_tag(buf, sizeof(buf), "localaddr", p)) {

626 av_strlcpy(localaddr, buf, sizeof(localaddr));

627 }

628 if (av_find_info_tag(buf, sizeof(buf), "sources", p)) {

629 if (parse_source_list(buf, include_sources, &num_include_sources,

630 FF_ARRAY_ELEMS(include_sources)))

631 goto fail;

632 }

633 if (av_find_info_tag(buf, sizeof(buf), "block", p)) {

634 if (parse_source_list(buf, exclude_sources, &num_exclude_sources,

635 FF_ARRAY_ELEMS(exclude_sources)))

636 goto fail;

637 }

638 if (!is_output && av_find_info_tag(buf, sizeof(buf), "timeout", p))

639 s->timeout = strtol(buf, NULL, 10);

640 if (is_output && av_find_info_tag(buf, sizeof(buf), "broadcast", p))

641 s->is_broadcast = strtol(buf, NULL, 10);

642 }

643 /* handling needed to support options picking from both AVOption and URL */

644 s->circular_buffer_size *= 188;

645 if (flags & AVIO_FLAG_WRITE) {

646 h->max_packet_size = s->packet_size;

647 } else {

648 h->max_packet_size = UDP_MAX_PKT_SIZE;

649 }

650 h->rw_timeout = s->timeout;

651

652 /* fill the dest addr */

653 av_url_split(NULL, 0, NULL, 0, hostname, sizeof(hostname), &port, NULL, 0, uri);

654

655 /* XXX: fix av_url_split */

656 if (hostname[0] == '0円' || hostname[0] == '?') {

657 /* only accepts null hostname if input */

658 if (!(flags & AVIO_FLAG_READ))

659 goto fail;

660 } else {

661 if (ff_udp_set_remote_url(h, uri) < 0)

662 goto fail;

663 }

664

665 if ((s->is_multicast || !s->local_port) && (h->flags & AVIO_FLAG_READ))

666 s->local_port = port;

667 udp_fd = udp_socket_create(s, &my_addr, &len, localaddr[0] ? localaddr : s->local_addr);

668 if (udp_fd < 0)

669 goto fail;

670

671 s->local_addr_storage=my_addr; //store for future multicast join

672

673 /* Follow the requested reuse option, unless it's multicast in which

674 * case enable reuse unless explicitly disabled.

675 */

676 if (s->reuse_socket || (s->is_multicast && !reuse_specified)) {

677 s->reuse_socket = 1;

678 if (setsockopt (udp_fd, SOL_SOCKET, SO_REUSEADDR, &(s->reuse_socket), sizeof(s->reuse_socket)) != 0)

679 goto fail;

680 }

681

682 if (s->is_broadcast) {

683 #ifdef SO_BROADCAST

684 if (setsockopt (udp_fd, SOL_SOCKET, SO_BROADCAST, &(s->is_broadcast), sizeof(s->is_broadcast)) != 0)

685 #endif

686 goto fail;

687 }

688

689 /* Set the checksum coverage for UDP-Lite (RFC 3828) for sending and receiving.

690 * The receiver coverage has to be less than or equal to the sender coverage.

691 * Otherwise, the receiver will drop all packets.

692 */

693 if (s->udplite_coverage) {

694 if (setsockopt (udp_fd, IPPROTO_UDPLITE, UDPLITE_SEND_CSCOV, &(s->udplite_coverage), sizeof(s->udplite_coverage)) != 0)

695 av_log(h, AV_LOG_WARNING, "socket option UDPLITE_SEND_CSCOV not available");

696

697 if (setsockopt (udp_fd, IPPROTO_UDPLITE, UDPLITE_RECV_CSCOV, &(s->udplite_coverage), sizeof(s->udplite_coverage)) != 0)

698 av_log(h, AV_LOG_WARNING, "socket option UDPLITE_RECV_CSCOV not available");

699 }

700

701 if (dscp >= 0) {

702 dscp <<= 2;

703 if (setsockopt (udp_fd, IPPROTO_IP, IP_TOS, &dscp, sizeof(dscp)) != 0)

704 goto fail;

705 }

706

707 /* If multicast, try binding the multicast address first, to avoid

708 * receiving UDP packets from other sources aimed at the same UDP

709 * port. This fails on windows. This makes sending to the same address

710 * using sendto() fail, so only do it if we're opened in read-only mode. */

711 if (s->is_multicast && !(h->flags & AVIO_FLAG_WRITE)) {

712 bind_ret = bind(udp_fd,(struct sockaddr *)&s->dest_addr, len);

713 }

714 /* bind to the local address if not multicast or if the multicast

715 * bind failed */

716 /* the bind is needed to give a port to the socket now */

717 if (bind_ret < 0 && bind(udp_fd,(struct sockaddr *)&my_addr, len) < 0) {

718 log_net_error(h, AV_LOG_ERROR, "bind failed");

719 goto fail;

720 }

721

722 len = sizeof(my_addr);

723 getsockname(udp_fd, (struct sockaddr *)&my_addr, &len);

724 s->local_port = udp_port(&my_addr, len);

725

726 if (s->is_multicast) {

727 if (h->flags & AVIO_FLAG_WRITE) {

728 /* output */

729 if (udp_set_multicast_ttl(udp_fd, s->ttl, (struct sockaddr *)&s->dest_addr) < 0)

730 goto fail;

731 }

732 if (h->flags & AVIO_FLAG_READ) {

733 /* input */

734 if (num_include_sources && num_exclude_sources) {

735 av_log(h, AV_LOG_ERROR, "Simultaneously including and excluding multicast sources is not supported\n");

736 goto fail;

737 }

738 if (num_include_sources) {

739 if (udp_set_multicast_sources(udp_fd, (struct sockaddr *)&s->dest_addr, s->dest_addr_len, include_sources, num_include_sources, 1) < 0)

740 goto fail;

741 } else {

742 if (udp_join_multicast_group(udp_fd, (struct sockaddr *)&s->dest_addr,(struct sockaddr *)&s->local_addr_storage) < 0)

743 goto fail;

744 }

745 if (num_exclude_sources) {

746 if (udp_set_multicast_sources(udp_fd, (struct sockaddr *)&s->dest_addr, s->dest_addr_len, exclude_sources, num_exclude_sources, 0) < 0)

747 goto fail;

748 }

749 }

750 }

751

752 if (is_output) {

753 /* limit the tx buf size to limit latency */

754 tmp = s->buffer_size;

755 if (setsockopt(udp_fd, SOL_SOCKET, SO_SNDBUF, &tmp, sizeof(tmp)) < 0) {

756 log_net_error(h, AV_LOG_ERROR, "setsockopt(SO_SNDBUF)");

757 goto fail;

758 }

759 } else {

760 /* set udp recv buffer size to the requested value (default 64K) */

761 tmp = s->buffer_size;

762 if (setsockopt(udp_fd, SOL_SOCKET, SO_RCVBUF, &tmp, sizeof(tmp)) < 0) {

763 log_net_error(h, AV_LOG_WARNING, "setsockopt(SO_RECVBUF)");

764 }

765 len = sizeof(tmp);

766 if (getsockopt(udp_fd, SOL_SOCKET, SO_RCVBUF, &tmp, &len) < 0) {

767 log_net_error(h, AV_LOG_WARNING, "getsockopt(SO_RCVBUF)");

768 } else {

769 av_log(h, AV_LOG_DEBUG, "end receive buffer size reported is %d\n", tmp);

770 if(tmp < s->buffer_size)

771 av_log(h, AV_LOG_WARNING, "attempted to set receive buffer to size %d but it only ended up set as %d", s->buffer_size, tmp);

772 }

773

774 /* make the socket non-blocking */

775 ff_socket_nonblock(udp_fd, 1);

776 }

777 if (s->is_connected) {

778 if (connect(udp_fd, (struct sockaddr *) &s->dest_addr, s->dest_addr_len)) {

779 log_net_error(h, AV_LOG_ERROR, "connect");

780 goto fail;

781 }

782 }

783

784 for (i = 0; i < num_include_sources; i++)

785 av_freep(&include_sources[i]);

786 for (i = 0; i < num_exclude_sources; i++)

787 av_freep(&exclude_sources[i]);

788

789 s->udp_fd = udp_fd;

790

791 #if HAVE_PTHREAD_CANCEL

792 if (!is_output && s->circular_buffer_size) {

793 int ret;

794

795 /* start the task going */

796 s->fifo = av_fifo_alloc(s->circular_buffer_size);

797 ret = pthread_mutex_init(&s->mutex, NULL);

798 if (ret != 0) {

799 av_log(h, AV_LOG_ERROR, "pthread_mutex_init failed : %s\n", strerror(ret));

800 goto fail;

801 }

802 ret = pthread_cond_init(&s->cond, NULL);

803 if (ret != 0) {

804 av_log(h, AV_LOG_ERROR, "pthread_cond_init failed : %s\n", strerror(ret));

805 goto cond_fail;

806 }

807 ret = pthread_create(&s->circular_buffer_thread, NULL, circular_buffer_task, h);

808 if (ret != 0) {

809 av_log(h, AV_LOG_ERROR, "pthread_create failed : %s\n", strerror(ret));

810 goto thread_fail;

811 }

812 s->thread_started = 1;

813 }

814 #endif

815

816 return 0;

817 #if HAVE_PTHREAD_CANCEL

818 thread_fail:

819 pthread_cond_destroy(&s->cond);

820 cond_fail:

821 pthread_mutex_destroy(&s->mutex);

822 #endif

823 fail:

824 if (udp_fd >= 0)

825 closesocket(udp_fd);

826 av_fifo_freep(&s->fifo);

827 for (i = 0; i < num_include_sources; i++)

828 av_freep(&include_sources[i]);

829 for (i = 0; i < num_exclude_sources; i++)

830 av_freep(&exclude_sources[i]);

831 return AVERROR(EIO);

832 }

833

834 static int udplite_open(URLContext *h, const char *uri, int flags)

835 {

836 UDPContext *s = h->priv_data;

837

838 // set default checksum coverage

839 s->udplite_coverage = UDP_HEADER_SIZE;

840

841 return udp_open(h, uri, flags);

842 }

843

844 static int udp_read(URLContext *h, uint8_t *buf, int size)

845 {

846 UDPContext *s = h->priv_data;

847 int ret;

848 #if HAVE_PTHREAD_CANCEL

849 int avail, nonblock = h->flags & AVIO_FLAG_NONBLOCK;

850

851 if (s->fifo) {

852 pthread_mutex_lock(&s->mutex);

853 do {

854 avail = av_fifo_size(s->fifo);

855 if (avail) { // >=size) {

856 uint8_t tmp[4];

857

858 av_fifo_generic_read(s->fifo, tmp, 4, NULL);

859 avail= AV_RL32(tmp);

860 if(avail > size){

861 av_log(h, AV_LOG_WARNING, "Part of datagram lost due to insufficient buffer size\n");

862 avail= size;

863 }

864

865 av_fifo_generic_read(s->fifo, buf, avail, NULL);

866 av_fifo_drain(s->fifo, AV_RL32(tmp) - avail);

867 pthread_mutex_unlock(&s->mutex);

868 return avail;

869 } else if(s->circular_buffer_error){

870 int err = s->circular_buffer_error;

871 pthread_mutex_unlock(&s->mutex);

872 return err;

873 } else if(nonblock) {

874 pthread_mutex_unlock(&s->mutex);

875 return AVERROR(EAGAIN);

876 }

877 else {

878 /* FIXME: using the monotonic clock would be better,

879 but it does not exist on all supported platforms. */

880 int64_t t = av_gettime() + 100000;

881 struct timespec tv = { .tv_sec = t / 1000000,

882 .tv_nsec = (t % 1000000) * 1000 };

883 if (pthread_cond_timedwait(&s->cond, &s->mutex, &tv) < 0) {

884 pthread_mutex_unlock(&s->mutex);

885 return AVERROR(errno == ETIMEDOUT ? EAGAIN : errno);

886 }

887 nonblock = 1;

888 }

889 } while( 1);

890 }

891 #endif

892

893 if (!(h->flags & AVIO_FLAG_NONBLOCK)) {

894 ret = ff_network_wait_fd(s->udp_fd, 0);

895 if (ret < 0)

896 return ret;

897 }

898 ret = recv(s->udp_fd, buf, size, 0);

899

900 return ret < 0 ? ff_neterrno() : ret;

901 }

902

903 static int udp_write(URLContext *h, const uint8_t *buf, int size)

904 {

905 UDPContext *s = h->priv_data;

906 int ret;

907

908 if (!(h->flags & AVIO_FLAG_NONBLOCK)) {

909 ret = ff_network_wait_fd(s->udp_fd, 1);

910 if (ret < 0)

911 return ret;

912 }

913

914 if (!s->is_connected) {

915 ret = sendto (s->udp_fd, buf, size, 0,

916 (struct sockaddr *) &s->dest_addr,

917 s->dest_addr_len);

918 } else

919 ret = send(s->udp_fd, buf, size, 0);

920

921 return ret < 0 ? ff_neterrno() : ret;

922 }

923

924 static int udp_close(URLContext *h)

925 {

926 UDPContext *s = h->priv_data;

927

928 if (s->is_multicast && (h->flags & AVIO_FLAG_READ))

929 udp_leave_multicast_group(s->udp_fd, (struct sockaddr *)&s->dest_addr,(struct sockaddr *)&s->local_addr_storage);

930 closesocket(s->udp_fd);

931 #if HAVE_PTHREAD_CANCEL

932 if (s->thread_started) {

933 int ret;

934 pthread_cancel(s->circular_buffer_thread);

935 ret = pthread_join(s->circular_buffer_thread, NULL);

936 if (ret != 0)

937 av_log(h, AV_LOG_ERROR, "pthread_join(): %s\n", strerror(ret));

938 pthread_mutex_destroy(&s->mutex);

939 pthread_cond_destroy(&s->cond);

940 }

941 #endif

942 av_fifo_freep(&s->fifo);

943 return 0;

944 }

945

946 URLProtocol ff_udp_protocol = {

947 .name = "udp",

948 .url_open = udp_open,

949 .url_read = udp_read,

950 .url_write = udp_write,

951 .url_close = udp_close,

952 .url_get_file_handle = udp_get_file_handle,

953 .priv_data_size = sizeof(UDPContext),

954 .priv_data_class = &udp_context_class,

955 .flags = URL_PROTOCOL_FLAG_NETWORK,

956 };

957

958 URLProtocol ff_udplite_protocol = {

959 .name = "udplite",

960 .url_open = udplite_open,

961 .url_read = udp_read,

962 .url_write = udp_write,

963 .url_close = udp_close,

964 .url_get_file_handle = udp_get_file_handle,

965 .priv_data_size = sizeof(UDPContext),

966 .priv_data_class = &udplite_context_class,

967 .flags = URL_PROTOCOL_FLAG_NETWORK,

968 };

Generated on Sun Mar 8 2015 02:35:13 for FFmpeg by doxygen 1.8.2