胡斌 / srs

转到一个项目
Toggle navigation Toggle navigation pinning
winlin
580cbb84 1 parent 435e02db

add threading.Lock for cherrypy, or performance suffer.

内嵌 并排对比
正在显示 1 个修改的文件 包含 198 行增加 和 167 行删除
@@ -368,7 +368,8 @@ class RESTServers(object): @@ -368,7 +368,8 @@ class RESTServers(object):
368 self.__nodes = [] 368 self.__nodes = []
369 369
370 self.__last_update = datetime.datetime.now(); 370 self.__last_update = datetime.datetime.now();
371 - server_ip = "192.168.1.142"; 371 +
  372 + self.__lock = threading.Lock()
372 373
373 def __get_node(self, device_id): 374 def __get_node(self, device_id):
374 for node in self.__nodes: 375 for node in self.__nodes:
@@ -415,28 +416,33 @@ class RESTServers(object): @@ -415,28 +416,33 @@ class RESTServers(object):
415 ''' 416 '''
416 def POST(self): 417 def POST(self):
417 enable_crossdomain() 418 enable_crossdomain()
418 -  
419 - req = cherrypy.request.body.read()  
420 - trace("post to nodes, req=%s"%(req)) 419 +
421 try: 420 try:
422 - json_req = json.loads(req)  
423 - except Exception, ex:  
424 - code = Error.system_parse_json  
425 - trace("parse the request to json failed, req=%s, ex=%s, code=%s"%(req, ex, code))  
426 - return json.dumps({"code":code, "data": None}) 421 + self.__lock.acquire()
  422 +
  423 + req = cherrypy.request.body.read()
  424 + trace("post to nodes, req=%s"%(req))
  425 + try:
  426 + json_req = json.loads(req)
  427 + except Exception, ex:
  428 + code = Error.system_parse_json
  429 + trace("parse the request to json failed, req=%s, ex=%s, code=%s"%(req, ex, code))
  430 + return json.dumps({"code":code, "data": None})
  431 +
  432 + device_id = json_req["device_id"]
  433 + node = self.__get_node(device_id)
  434 + if node is None:
  435 + node = ArmServer()
  436 + self.__nodes.append(node)
  437 +
  438 + node.ip = json_req["ip"]
  439 + node.device_id = device_id
  440 + node.public_ip = cherrypy.request.remote.ip
  441 + node.heartbeat = time.time()
427 442
428 - device_id = json_req["device_id"]  
429 - node = self.__get_node(device_id)  
430 - if node is None:  
431 - node = ArmServer()  
432 - self.__nodes.append(node)  
433 -  
434 - node.ip = json_req["ip"]  
435 - node.device_id = device_id  
436 - node.public_ip = cherrypy.request.remote.ip  
437 - node.heartbeat = time.time()  
438 -  
439 - return json.dumps({"code":Error.success, "data": {"id":node.id}}) 443 + return json.dumps({"code":Error.success, "data": {"id":node.id}})
  444 + finally:
  445 + self.__lock.release()
440 446
441 ''' 447 '''
442 id canbe: 448 id canbe:
@@ -455,58 +461,63 @@ class RESTServers(object): @@ -455,58 +461,63 @@ class RESTServers(object):
455 def GET(self, id=None, action="play", stream="live/livestream", index=None, local="false", device_id=None): 461 def GET(self, id=None, action="play", stream="live/livestream", index=None, local="false", device_id=None):
456 enable_crossdomain() 462 enable_crossdomain()
457 463
458 - self.__refresh_nodes()  
459 - data = self.__json_dump_nodes(self.__nodes)  
460 -  
461 - server_ip = "demo.chnvideo.com"  
462 - ip = cherrypy.request.remote.ip  
463 - if type is not None:  
464 - peers = self.__get_peers_for_play(device_id)  
465 - if len(peers) > 0:  
466 - server_ip = self.__select_peer(peers, device_id) 464 + try:
  465 + self.__lock.acquire()
467 466
468 - # demo, srs meeting urls.  
469 - if id == "meeting":  
470 - if index is None:  
471 - url = "http://%s:8085"%(server_ip)  
472 - elif local == "true":  
473 - url = "http://%s:8085/api/v1/servers?id=%s&index=%s&local=false"%(server_ip, id, index)  
474 - else:  
475 - rtmp_url = root.api.v1.chats.get_url_by_index(index)  
476 - if rtmp_url is None:  
477 - return "meeting stream not found"  
478 - urls = rtmp_url.replace("...vhost...", "?vhost=").replace("rtmp://", "").split("/")  
479 - hls_url = "http://%s:8080/%s/%s.m3u8"%(urls[0].strip(":19350").strip(":1935"), urls[1].split("?")[0], urls[2])  
480 - return self.__generate_hls(hls_url)  
481 - # raspberry-pi urls.  
482 - elif id == "ingest" or id == "pi":  
483 - if action == "play":  
484 - url = "http://%s:8080/%s.html"%(server_ip, stream)  
485 - elif action == "rtmp":  
486 - url = "../../players/srs_player.html?server=%s&vhost=%s&app=%s&stream=%s&autostart=true"%(server_ip, server_ip, stream.split("/")[0], stream.split("/")[1])  
487 - elif action == "hls":  
488 - hls_url = "http://%s:8080/%s.m3u8"%(server_ip, stream);  
489 - if stream.startswith("http://"):  
490 - hls_url = stream;  
491 - return self.__generate_hls(hls_url.replace(".m3u8.m3u8", ".m3u8")) 467 + self.__refresh_nodes()
  468 + data = self.__json_dump_nodes(self.__nodes)
  469 +
  470 + server_ip = "demo.chnvideo.com"
  471 + ip = cherrypy.request.remote.ip
  472 + if type is not None:
  473 + peers = self.__get_peers_for_play(device_id)
  474 + if len(peers) > 0:
  475 + server_ip = self.__select_peer(peers, device_id)
  476 +
  477 + # demo, srs meeting urls.
  478 + if id == "meeting":
  479 + if index is None:
  480 + url = "http://%s:8085"%(server_ip)
  481 + elif local == "true":
  482 + url = "http://%s:8085/api/v1/servers?id=%s&index=%s&local=false"%(server_ip, id, index)
  483 + else:
  484 + rtmp_url = root.api.v1.chats.get_url_by_index(index)
  485 + if rtmp_url is None:
  486 + return "meeting stream not found"
  487 + urls = rtmp_url.replace("...vhost...", "?vhost=").replace("rtmp://", "").split("/")
  488 + hls_url = "http://%s:8080/%s/%s.m3u8"%(urls[0].strip(":19350").strip(":1935"), urls[1].split("?")[0], urls[2])
  489 + return self.__generate_hls(hls_url)
  490 + # raspberry-pi urls.
  491 + elif id == "ingest" or id == "pi":
  492 + if action == "play":
  493 + url = "http://%s:8080/%s.html"%(server_ip, stream)
  494 + elif action == "rtmp":
  495 + url = "../../players/srs_player.html?server=%s&vhost=%s&app=%s&stream=%s&autostart=true"%(server_ip, server_ip, stream.split("/")[0], stream.split("/")[1])
  496 + elif action == "hls":
  497 + hls_url = "http://%s:8080/%s.m3u8"%(server_ip, stream);
  498 + if stream.startswith("http://"):
  499 + hls_url = stream;
  500 + return self.__generate_hls(hls_url.replace(".m3u8.m3u8", ".m3u8"))
  501 + else:
  502 + url = "http://%s:8080/api/v1/versions"%(server_ip)
  503 + elif id == "gslb":
  504 + return json.dumps({"code":Error.success, "data": {
  505 + "edge":server_ip, "client":ip,
  506 + "peers":self.__json_dump_nodes(peers),
  507 + "streams": {
  508 + "hls-livestream-sales": "http://demo.chnvideo.com:8085/api/v1/servers?id=ingest&action=hls&device_id=chnvideo-sales-arm&stream=live/livestream",
  509 + "hls-cztv-sales": "http://demo.chnvideo.com:8085/api/v1/servers?id=ingest&action=hls&device_id=chnvideo-sales-arm&stream=live/rtmp_cztv01-sd",
  510 + "hls-livestream-dev": "http://demo.chnvideo.com:8085/api/v1/servers?id=ingest&action=hls&device_id=chnvideo-dev-arm&stream=live/livestream",
  511 + "hls-cztv-dev": "http://demo.chnvideo.com:8085/api/v1/servers?id=ingest&action=hls&device_id=chnvideo-dev-arm&stream=live/rtmp_cztv01-sd"
  512 + }
  513 + }})
  514 + # others, default.
492 else: 515 else:
493 - url = "http://%s:8080/api/v1/versions"%(server_ip)  
494 - elif id == "gslb":  
495 - return json.dumps({"code":Error.success, "data": {  
496 - "edge":server_ip, "client":ip,  
497 - "peers":self.__json_dump_nodes(peers),  
498 - "streams": {  
499 - "hls-livestream-sales": "http://demo.chnvideo.com:8085/api/v1/servers?id=ingest&action=hls&device_id=chnvideo-sales-arm&stream=live/livestream",  
500 - "hls-cztv-sales": "http://demo.chnvideo.com:8085/api/v1/servers?id=ingest&action=hls&device_id=chnvideo-sales-arm&stream=live/rtmp_cztv01-sd",  
501 - "hls-livestream-dev": "http://demo.chnvideo.com:8085/api/v1/servers?id=ingest&action=hls&device_id=chnvideo-dev-arm&stream=live/livestream",  
502 - "hls-cztv-dev": "http://demo.chnvideo.com:8085/api/v1/servers?id=ingest&action=hls&device_id=chnvideo-dev-arm&stream=live/rtmp_cztv01-sd"  
503 - }  
504 - }})  
505 - # others, default.  
506 - else:  
507 - return json.dumps(data)  
508 - #return "id=%s, action=%s, stream=%s, url=%s, index=%s, local=%s"%(id, action, stream, url, index, local)  
509 - raise cherrypy.HTTPRedirect(url) 516 + return json.dumps(data)
  517 + #return "id=%s, action=%s, stream=%s, url=%s, index=%s, local=%s"%(id, action, stream, url, index, local)
  518 + raise cherrypy.HTTPRedirect(url)
  519 + finally:
  520 + self.__lock.release()
510 521
511 def DELETE(self, id): 522 def DELETE(self, id):
512 enable_crossdomain() 523 enable_crossdomain()
@@ -577,6 +588,10 @@ class RESTNodes(object): @@ -577,6 +588,10 @@ class RESTNodes(object):
577 588
578 def __init__(self): 589 def __init__(self):
579 self.__nodes = [] 590 self.__nodes = []
  591 + # @remark, if there is shared data, such as the self.__nodes,
  592 + # we must use lock for cherrypy, or the cpu of cherrypy will high
  593 + # and performance suffer.
  594 + self.__lock = threading.Lock()
580 595
581 def __get_node(self, id): 596 def __get_node(self, id):
582 for node in self.__nodes: 597 for node in self.__nodes:
@@ -629,113 +644,128 @@ class RESTNodes(object): @@ -629,113 +644,128 @@ class RESTNodes(object):
629 def GET(self, type=None, format=None, origin=None, vhost=None, port=None, stream=None, node_id=None): 644 def GET(self, type=None, format=None, origin=None, vhost=None, port=None, stream=None, node_id=None):
630 enable_crossdomain() 645 enable_crossdomain()
631 646
632 - self.__refresh_nodes()  
633 - data = self.__json_dump_nodes(self.__nodes)  
634 -  
635 - ip = cherrypy.request.remote.ip  
636 - if type is not None:  
637 - server = origin  
638 - peers = self.__get_peers_for_play(ip)  
639 - if len(peers) > 0:  
640 - server = self.__select_peer(peers, ip)  
641 - if type == "hls":  
642 - hls_url = "http://%s:%s/%s.m3u8"%(server, port, stream)  
643 - hls_url = hls_url.replace(".m3u8.m3u8", ".m3u8")  
644 - if format == "html":  
645 - return SrsUtility().hls_html(hls_url)  
646 - else:  
647 - #return hls_url  
648 - raise cherrypy.HTTPRedirect(hls_url)  
649 - elif type == "rtmp":  
650 - rtmp_url = "rtmp://%s:%s/%s?vhost=%s/%s"%(server, port, stream.split("/")[0], vhost, stream.split("/")[1])  
651 - if format == "html":  
652 - html = "%s?server=%s&port=%s&vhost=%s&app=%s&stream=%s&autostart=true"%(  
653 - "http://demo.chnvideo.com:8085/srs/trunk/research/players/srs_player.html",  
654 - server, port, vhost, stream.split("/")[0], stream.split("/")[1])  
655 - #return html  
656 - raise cherrypy.HTTPRedirect(html)  
657 - return rtmp_url  
658 - elif type == "gslb":  
659 - return json.dumps({"code":Error.success, "data": {  
660 - "edge":server, "client":ip,  
661 - "peers":self.__json_dump_nodes(peers),  
662 - "streams": {  
663 - "hls-cztv-html": "http://demo.chnvideo.com:8085/api/v1/nodes?type=hls&format=html&origin=demo.chnvideo.com&port=8080&stream=live/rtmp_cztv01-sd",  
664 - "hls-cztv-m3u8": "http://demo.chnvideo.com:8085/api/v1/nodes?type=hls&format=m3u8&origin=demo.chnvideo.com&port=8080&stream=live/rtmp_cztv01-sd",  
665 - "rtmp-cztv-html": "http://demo.chnvideo.com:8085/api/v1/nodes?type=rtmp&format=html&origin=demo.chnvideo.com&vhost=__defaultVhost__&port=1935&stream=live/rtmp_cztv01-sd",  
666 - "hls-livestream-html": "http://demo.chnvideo.com:8085/api/v1/nodes?type=hls&format=html&origin=demo.chnvideo.com&port=8080&stream=live/livestream",  
667 - "hls-livestream-m3u8": "http://demo.chnvideo.com:8085/api/v1/nodes?type=hls&format=m3u8&origin=demo.chnvideo.com&port=8080&stream=live/livestream",  
668 - "rtmp-livestream-html": "http://demo.chnvideo.com:8085/api/v1/nodes?type=rtmp&format=html&origin=demo.chnvideo.com&vhost=demo.srs.com&port=1935&stream=live/livestream",  
669 - "apk": "http://demo.chnvideo.com/android.srs.apk"  
670 - }  
671 - }}) 647 + try:
  648 + self.__lock.acquire()
672 649
673 - return json.dumps({"code":Error.success, "data": data}) 650 + self.__refresh_nodes()
  651 + data = self.__json_dump_nodes(self.__nodes)
  652 +
  653 + ip = cherrypy.request.remote.ip
  654 + if type is not None:
  655 + server = origin
  656 + peers = self.__get_peers_for_play(ip)
  657 + if len(peers) > 0:
  658 + server = self.__select_peer(peers, ip)
  659 + if type == "hls":
  660 + hls_url = "http://%s:%s/%s.m3u8"%(server, port, stream)
  661 + hls_url = hls_url.replace(".m3u8.m3u8", ".m3u8")
  662 + if format == "html":
  663 + return SrsUtility().hls_html(hls_url)
  664 + else:
  665 + #return hls_url
  666 + raise cherrypy.HTTPRedirect(hls_url)
  667 + elif type == "rtmp":
  668 + rtmp_url = "rtmp://%s:%s/%s?vhost=%s/%s"%(server, port, stream.split("/")[0], vhost, stream.split("/")[1])
  669 + if format == "html":
  670 + html = "%s?server=%s&port=%s&vhost=%s&app=%s&stream=%s&autostart=true"%(
  671 + "http://demo.chnvideo.com:8085/srs/trunk/research/players/srs_player.html",
  672 + server, port, vhost, stream.split("/")[0], stream.split("/")[1])
  673 + #return html
  674 + raise cherrypy.HTTPRedirect(html)
  675 + return rtmp_url
  676 + elif type == "gslb":
  677 + return json.dumps({"code":Error.success, "data": {
  678 + "edge":server, "client":ip,
  679 + "peers":self.__json_dump_nodes(peers),
  680 + "streams": {
  681 + "hls-cztv-html": "http://demo.chnvideo.com:8085/api/v1/nodes?type=hls&format=html&origin=demo.chnvideo.com&port=8080&stream=live/rtmp_cztv01-sd",
  682 + "hls-cztv-m3u8": "http://demo.chnvideo.com:8085/api/v1/nodes?type=hls&format=m3u8&origin=demo.chnvideo.com&port=8080&stream=live/rtmp_cztv01-sd",
  683 + "rtmp-cztv-html": "http://demo.chnvideo.com:8085/api/v1/nodes?type=rtmp&format=html&origin=demo.chnvideo.com&vhost=__defaultVhost__&port=1935&stream=live/rtmp_cztv01-sd",
  684 + "hls-livestream-html": "http://demo.chnvideo.com:8085/api/v1/nodes?type=hls&format=html&origin=demo.chnvideo.com&port=8080&stream=live/livestream",
  685 + "hls-livestream-m3u8": "http://demo.chnvideo.com:8085/api/v1/nodes?type=hls&format=m3u8&origin=demo.chnvideo.com&port=8080&stream=live/livestream",
  686 + "rtmp-livestream-html": "http://demo.chnvideo.com:8085/api/v1/nodes?type=rtmp&format=html&origin=demo.chnvideo.com&vhost=demo.srs.com&port=1935&stream=live/livestream",
  687 + "apk": "http://demo.chnvideo.com/android.srs.apk"
  688 + }
  689 + }})
  690 +
  691 + return json.dumps({"code":Error.success, "data": data})
  692 + finally:
  693 + self.__lock.release()
674 694
675 def PUT(self): 695 def PUT(self):
676 enable_crossdomain() 696 enable_crossdomain()
677 -  
678 - req = cherrypy.request.body.read()  
679 - trace("put to nodes, req=%s"%(req))  
680 - try:  
681 - json_req = json.loads(req)  
682 - except Exception, ex:  
683 - code = Error.system_parse_json  
684 - trace("parse the request to json failed, req=%s, ex=%s, code=%s"%(req, ex, code))  
685 - return json.dumps({"code":code, "data": None})  
686 697
687 - id = str(json_req["id"])  
688 - node = self.__get_node(id)  
689 - if node is None:  
690 - code = Error.cdn_node_not_exists  
691 - trace("cdn node not exists, req=%s, id=%s, code=%s"%(req, id, code))  
692 - return json.dumps({"code":code, "data": None})  
693 -  
694 - node.heartbeat = time.time()  
695 - node.srs_status = str(json_req["srs_status"])  
696 - node.ip = str(json_req["ip"])  
697 - if "origin" in json_req:  
698 - node.origin = str(json_req["origin"]);  
699 - node.public_ip = cherrypy.request.remote.ip  
700 - # reset if restart.  
701 - if node.srs_status != "running":  
702 - node.clients = 0 698 + try:
  699 + self.__lock.acquire()
  700 +
  701 + req = cherrypy.request.body.read()
  702 + trace("put to nodes, req=%s"%(req))
  703 + try:
  704 + json_req = json.loads(req)
  705 + except Exception, ex:
  706 + code = Error.system_parse_json
  707 + trace("parse the request to json failed, req=%s, ex=%s, code=%s"%(req, ex, code))
  708 + return json.dumps({"code":code, "data": None})
703 709
704 - self.__refresh_nodes()  
705 - peers = self.__get_peers(node)  
706 - peers_data = self.__json_dump_nodes(peers) 710 + id = str(json_req["id"])
  711 + node = self.__get_node(id)
  712 + if node is None:
  713 + code = Error.cdn_node_not_exists
  714 + trace("cdn node not exists, req=%s, id=%s, code=%s"%(req, id, code))
  715 + return json.dumps({"code":code, "data": None})
707 716
708 - res = json.dumps({"code":Error.success, "data": {"id":node.id, "peers":peers_data}})  
709 - trace(res)  
710 - return res 717 + node.heartbeat = time.time()
  718 + node.srs_status = str(json_req["srs_status"])
  719 + node.ip = str(json_req["ip"])
  720 + if "origin" in json_req:
  721 + node.origin = str(json_req["origin"]);
  722 + node.public_ip = cherrypy.request.remote.ip
  723 + # reset if restart.
  724 + if node.srs_status != "running":
  725 + node.clients = 0
  726 +
  727 + self.__refresh_nodes()
  728 + peers = self.__get_peers(node)
  729 + peers_data = self.__json_dump_nodes(peers)
  730 +
  731 + res = json.dumps({"code":Error.success, "data": {"id":node.id, "peers":peers_data}})
  732 + trace(res)
  733 + return res
  734 + finally:
  735 + self.__lock.release()
711 736
712 def POST(self): 737 def POST(self):
713 enable_crossdomain() 738 enable_crossdomain()
714 -  
715 - req = cherrypy.request.body.read()  
716 - trace("post to nodes, req=%s"%(req)) 739 +
717 try: 740 try:
718 - json_req = json.loads(req)  
719 - except Exception, ex:  
720 - code = Error.system_parse_json  
721 - trace("parse the request to json failed, req=%s, ex=%s, code=%s"%(req, ex, code))  
722 - return json.dumps({"code":code, "data": None})  
723 -  
724 - node = CdnNode()  
725 - node.ip = str(json_req["ip"]);  
726 - node.os = str(json_req["os"]);  
727 - if "origin" in json_req:  
728 - node.origin = str(json_req["origin"]);  
729 - node.srs_status = str(json_req["srs_status"])  
730 - self.__nodes.append(node) 741 + self.__lock.acquire()
  742 +
  743 + req = cherrypy.request.body.read()
  744 + trace("post to nodes, req=%s"%(req))
  745 + try:
  746 + json_req = json.loads(req)
  747 + except Exception, ex:
  748 + code = Error.system_parse_json
  749 + trace("parse the request to json failed, req=%s, ex=%s, code=%s"%(req, ex, code))
  750 + return json.dumps({"code":code, "data": None})
  751 +
  752 + node = CdnNode()
  753 + node.ip = str(json_req["ip"]);
  754 + node.os = str(json_req["os"]);
  755 + if "origin" in json_req:
  756 + node.origin = str(json_req["origin"]);
  757 + node.srs_status = str(json_req["srs_status"])
  758 + self.__nodes.append(node)
  759 +
  760 + self.__refresh_nodes()
  761 + peers = self.__get_peers(node)
  762 + peers_data = self.__json_dump_nodes(peers)
731 763
732 - self.__refresh_nodes()  
733 - peers = self.__get_peers(node)  
734 - peers_data = self.__json_dump_nodes(peers)  
735 -  
736 - res = json.dumps({"code":Error.success, "data": {"id":node.id, "peers":peers_data}})  
737 - trace(res)  
738 - return res 764 + res = json.dumps({"code":Error.success, "data": {"id":node.id, "peers":peers_data}})
  765 + trace(res)
  766 + return res
  767 + finally:
  768 + self.__lock.release()
739 769
740 def OPTIONS(self, *args, **kwargs): 770 def OPTIONS(self, *args, **kwargs):
741 enable_crossdomain() 771 enable_crossdomain()
@@ -966,7 +996,8 @@ conf = { @@ -966,7 +996,8 @@ conf = {
966 'server.socket_port': port, 996 'server.socket_port': port,
967 'tools.encode.on': True, 997 'tools.encode.on': True,
968 'tools.staticdir.on': True, 998 'tools.staticdir.on': True,
969 - 'tools.encode.encoding': "utf-8" 999 + 'tools.encode.encoding': "utf-8",
  1000 + 'server.thread_pool': 2, # single thread server.
970 }, 1001 },
971 '/': { 1002 '/': {
972 'tools.staticdir.dir': static_dir, 1003 'tools.staticdir.dir': static_dir,