chore: import upstream snapshot with attribution
This commit is contained in:
wehub-resource-sync
@@ -0,0 +1,731 @@
|
||||
# coding: utf-8
|
||||
from __future__ import print_function, unicode_literals
|
||||
|
||||
import hashlib
|
||||
import math
|
||||
import os
|
||||
import re
|
||||
import socket
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
import queue
|
||||
|
||||
from .__init__ import ANYWIN, CORES, EXE, MACOS, PY2, TYPE_CHECKING, EnvParams, unicode
|
||||
|
||||
try:
|
||||
MNFE = ModuleNotFoundError
|
||||
except:
|
||||
MNFE = ImportError
|
||||
|
||||
try:
|
||||
import jinja2
|
||||
except MNFE:
|
||||
if EXE:
|
||||
raise
|
||||
|
||||
print(
|
||||
"""\033[1;31m
|
||||
you do not have jinja2 installed,\033[33m
|
||||
choose one of these:\033[0m
|
||||
* apt install python-jinja2
|
||||
* {} -m pip install --user jinja2
|
||||
* (try another python version, if you have one)
|
||||
* (try copyparty.sfx instead)
|
||||
""".format(
|
||||
sys.executable
|
||||
)
|
||||
)
|
||||
sys.exit(1)
|
||||
except SyntaxError:
|
||||
if EXE:
|
||||
raise
|
||||
|
||||
print(
|
||||
"""\033[1;31m
|
||||
your jinja2 version is incompatible with your python version;\033[33m
|
||||
please try to replace it with an older version:\033[0m
|
||||
* {} -m pip install --user jinja2==2.11.3
|
||||
* (try another python version, if you have one)
|
||||
* (try copyparty.sfx instead)
|
||||
""".format(
|
||||
sys.executable
|
||||
)
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
from .authsrv import LEELOO_DALLAS
|
||||
from .httpconn import HttpConn
|
||||
from .ico import Ico
|
||||
from .metrics import Metrics
|
||||
from .mtag import HAVE_FFMPEG
|
||||
from .sutil import gfilter2
|
||||
from .th_cli import ThumbCli
|
||||
from .th_srv import HAVE_PIL, HAVE_VIPS
|
||||
from .u2idx import U2idx
|
||||
from .util import (
|
||||
E_SCK,
|
||||
FHC,
|
||||
CachedDict,
|
||||
Daemon,
|
||||
Garda,
|
||||
Magician,
|
||||
Netdev,
|
||||
NetMap,
|
||||
build_netmap,
|
||||
has_resource,
|
||||
ipnorm,
|
||||
load_ipr,
|
||||
load_ipu,
|
||||
load_resource,
|
||||
min_ex,
|
||||
shut_socket,
|
||||
spack,
|
||||
start_log_thrs,
|
||||
start_stackmon,
|
||||
ub64enc,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .authsrv import VFS
|
||||
from .broker_util import BrokerCli
|
||||
from .ssdp import SSDPr
|
||||
|
||||
if True: # pylint: disable=using-constant-test
|
||||
from typing import Any, Optional
|
||||
|
||||
if PY2:
|
||||
range = xrange # type: ignore
|
||||
|
||||
if not hasattr(socket, "AF_UNIX"):
|
||||
setattr(socket, "AF_UNIX", -9001)
|
||||
|
||||
|
||||
def load_jinja2_resource(E: EnvParams, name: str):
|
||||
with load_resource(E, "web/" + name, "r") as f:
|
||||
return f.read()
|
||||
|
||||
|
||||
class HttpSrv(object):
|
||||
"""
|
||||
handles incoming connections using HttpConn to process http,
|
||||
relying on MpSrv for performance (HttpSrv is just plain threads)
|
||||
"""
|
||||
|
||||
def __init__(self, broker: "BrokerCli", nid: Optional[int]) -> None:
|
||||
self.broker = broker
|
||||
self.nid = nid
|
||||
self.args = broker.args
|
||||
self.E: EnvParams = self.args.E
|
||||
self.log = broker.log
|
||||
self.asrv = broker.asrv
|
||||
|
||||
# redefine in case of multiprocessing
|
||||
socket.setdefaulttimeout(120)
|
||||
|
||||
self.t0 = time.time()
|
||||
nsuf = "-n{}-i{:x}".format(nid, os.getpid()) if nid else ""
|
||||
self.magician = Magician()
|
||||
self.nm = NetMap([], [])
|
||||
self.ssdp: Optional["SSDPr"] = None
|
||||
self.gpwd = Garda(self.args.ban_pw)
|
||||
self.gpwc = Garda(self.args.ban_pwc)
|
||||
self.g404 = Garda(self.args.ban_404)
|
||||
self.g403 = Garda(self.args.ban_403)
|
||||
self.g422 = Garda(self.args.ban_422, False)
|
||||
self.gmal = Garda(self.args.ban_422)
|
||||
self.gurl = Garda(self.args.ban_url)
|
||||
self.bans: dict[str, int] = {}
|
||||
self.aclose: dict[str, int] = {}
|
||||
|
||||
self.thumbcli: Optional[ThumbCli] = None
|
||||
self.ico: Ico = Ico(self.args)
|
||||
|
||||
dli: dict[str, tuple[float, int, "VFS", str, str]] = {} # info
|
||||
dls: dict[str, tuple[float, int]] = {} # state
|
||||
self.dli = self.tdli = dli
|
||||
self.dls = self.tdls = dls
|
||||
self.iiam = '<img src="%s.cpr/w/iiam.gif?cache=i" />' % (self.args.SRS,)
|
||||
|
||||
self.bound: set[tuple[str, int]] = set()
|
||||
self.name = "hsrv" + nsuf
|
||||
self.mutex = threading.Lock()
|
||||
self.u2mutex = threading.Lock()
|
||||
self.bad_ver = False
|
||||
self.stopping = False
|
||||
|
||||
self.tp_nthr = 0 # actual
|
||||
self.tp_ncli = 0 # fading
|
||||
self.tp_time = 0.0 # latest worker collect
|
||||
self.tp_q: Optional[queue.LifoQueue[Any]] = (
|
||||
None if self.args.no_htp else queue.LifoQueue()
|
||||
)
|
||||
self.t_periodic: Optional[threading.Thread] = None
|
||||
|
||||
self.u2fh = FHC()
|
||||
self.u2sc: dict[str, tuple[int, "hashlib._Hash"]] = {}
|
||||
self.pipes = CachedDict(0.2)
|
||||
self.metrics = Metrics(self)
|
||||
self.nreq = 0
|
||||
self.nsus = 0
|
||||
self.nban = 0
|
||||
self.srvs: list[socket.socket] = []
|
||||
self.ncli = 0 # exact
|
||||
self.clients: set[HttpConn] = set() # laggy
|
||||
self.nclimax = 0
|
||||
self.cb_ts = 0.0
|
||||
self.cb_v = ""
|
||||
|
||||
self.u2idx_free: dict[str, U2idx] = {}
|
||||
self.u2idx_n = 0
|
||||
|
||||
assert jinja2 # type: ignore # !rm
|
||||
env = jinja2.Environment()
|
||||
env.loader = jinja2.FunctionLoader(lambda f: load_jinja2_resource(self.E, f))
|
||||
jn = [
|
||||
"browser",
|
||||
"browser2",
|
||||
"cf",
|
||||
"idp",
|
||||
"md",
|
||||
"mde",
|
||||
"msg",
|
||||
"rups",
|
||||
"shares",
|
||||
"splash",
|
||||
"svcs",
|
||||
]
|
||||
self.j2 = {x: env.get_template(x + ".html") for x in jn}
|
||||
self.j2["opds"] = env.get_template("opds.xml")
|
||||
self.j2["opds_osd"] = env.get_template("opds_osd.xml")
|
||||
self.prism = has_resource(self.E, "web/deps/prism.js.gz")
|
||||
|
||||
if self.args.ipu:
|
||||
self.ipu_iu, self.ipu_nm = load_ipu(self.log, self.args.ipu)
|
||||
else:
|
||||
self.ipu_iu = self.ipu_nm = None
|
||||
|
||||
if self.args.ipr:
|
||||
self.ipr = load_ipr(self.log, self.args.ipr)
|
||||
else:
|
||||
self.ipr = None
|
||||
|
||||
self.ipa_nm = build_netmap(self.args.ipa)
|
||||
self.ipar_nm = build_netmap(self.args.ipar)
|
||||
self.xff_nm = build_netmap(self.args.xff_src)
|
||||
self.xff_lan = build_netmap("lan")
|
||||
|
||||
self.mallow = "GET HEAD POST PUT DELETE OPTIONS".split()
|
||||
if not self.args.no_dav:
|
||||
zs = "PROPFIND PROPPATCH LOCK UNLOCK MKCOL COPY MOVE"
|
||||
self.mallow += zs.split()
|
||||
|
||||
if self.args.zs:
|
||||
from .ssdp import SSDPr
|
||||
|
||||
self.ssdp = SSDPr(broker)
|
||||
|
||||
if self.tp_q:
|
||||
self.start_threads(4)
|
||||
|
||||
if nid:
|
||||
self.tdli = {}
|
||||
self.tdls = {}
|
||||
|
||||
if self.args.stackmon:
|
||||
start_stackmon(self.args.stackmon, nid)
|
||||
|
||||
if self.args.log_thrs:
|
||||
start_log_thrs(self.log, self.args.log_thrs, nid)
|
||||
|
||||
if (HAVE_PIL or HAVE_VIPS or HAVE_FFMPEG) and not self.args.no_thumb:
|
||||
Daemon(self.post_init, "hsrv-init2")
|
||||
|
||||
def post_init(self) -> None:
|
||||
try:
|
||||
x = self.broker.ask("thumbsrv.getcfg")
|
||||
self.thumbcli = ThumbCli(self, x.get())
|
||||
except:
|
||||
pass
|
||||
|
||||
def set_th_cfg(self, c: dict[str, set[str]], opts: tuple[bool]) -> None:
|
||||
self.args.th_no_jxl = opts[0]
|
||||
self.thumbcli = ThumbCli(self, c)
|
||||
|
||||
def set_bad_ver(self) -> None:
|
||||
self.bad_ver = True
|
||||
|
||||
def set_netdevs(self, netdevs: dict[str, Netdev]) -> None:
|
||||
ips = set()
|
||||
for ip, _ in self.bound:
|
||||
ips.add(ip)
|
||||
|
||||
self.nm = NetMap(list(ips), list(netdevs))
|
||||
|
||||
def start_threads(self, n: int) -> None:
|
||||
self.tp_nthr += n
|
||||
if self.args.log_htp:
|
||||
self.log(self.name, "workers += {} = {}".format(n, self.tp_nthr), 6)
|
||||
|
||||
for _ in range(n):
|
||||
Daemon(self.thr_poolw, self.name + "-poolw")
|
||||
|
||||
def stop_threads(self, n: int) -> None:
|
||||
self.tp_nthr -= n
|
||||
if self.args.log_htp:
|
||||
self.log(self.name, "workers -= {} = {}".format(n, self.tp_nthr), 6)
|
||||
|
||||
assert self.tp_q # !rm
|
||||
for _ in range(n):
|
||||
self.tp_q.put(None)
|
||||
|
||||
def periodic(self) -> None:
|
||||
while True:
|
||||
time.sleep(2 if self.tp_ncli or self.ncli else 10)
|
||||
with self.u2mutex, self.mutex:
|
||||
self.u2fh.clean()
|
||||
if self.tp_q:
|
||||
self.tp_ncli = max(self.ncli, self.tp_ncli - 2)
|
||||
if self.tp_nthr > self.tp_ncli + 8:
|
||||
self.stop_threads(4)
|
||||
|
||||
if not self.ncli and not self.u2fh.cache and self.tp_nthr <= 8:
|
||||
self.t_periodic = None
|
||||
return
|
||||
|
||||
def listen(self, sck: socket.socket, nlisteners: int) -> None:
|
||||
tcp = sck.family != socket.AF_UNIX
|
||||
|
||||
if self.args.j != 1:
|
||||
# lost in the pickle; redefine
|
||||
if not ANYWIN or self.args.reuseaddr:
|
||||
sck.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
|
||||
if tcp:
|
||||
sck.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
|
||||
|
||||
sck.settimeout(None) # < does not inherit, ^ opts above do
|
||||
|
||||
if tcp:
|
||||
ip, port = sck.getsockname()[:2]
|
||||
else:
|
||||
ip = re.sub(r"\.[0-9]+$", "", sck.getsockname().split("/")[-1])
|
||||
port = 0
|
||||
|
||||
self.srvs.append(sck)
|
||||
self.bound.add((ip, port))
|
||||
self.nclimax = math.ceil(self.args.nc * 1.0 / nlisteners)
|
||||
Daemon(
|
||||
self.thr_listen,
|
||||
"httpsrv-n{}-listen-{}-{}".format(self.nid or "0", ip, port),
|
||||
(sck,),
|
||||
)
|
||||
|
||||
def thr_listen(self, srv_sck: socket.socket) -> None:
|
||||
"""listens on a shared tcp server"""
|
||||
fno = srv_sck.fileno()
|
||||
if srv_sck.family == socket.AF_UNIX:
|
||||
ip = re.sub(r"\.[0-9]+$", "", srv_sck.getsockname())
|
||||
msg = "subscribed @ %s f%d p%d" % (ip, fno, os.getpid())
|
||||
ip = ip.split("/")[-1]
|
||||
port = 0
|
||||
tcp = False
|
||||
else:
|
||||
tcp = True
|
||||
ip, port = srv_sck.getsockname()[:2]
|
||||
hip = "[%s]" % (ip,) if ":" in ip else ip
|
||||
msg = "subscribed @ %s:%d f%d p%d" % (hip, port, fno, os.getpid())
|
||||
|
||||
self.log(self.name, msg)
|
||||
|
||||
Daemon(self.broker.say, "sig-hsrv-up1", ("cb_httpsrv_up",))
|
||||
|
||||
saddr = ("", 0) # fwd-decl for `except TypeError as ex:`
|
||||
|
||||
while not self.stopping:
|
||||
if self.args.log_conn:
|
||||
self.log(self.name, "|%sC-ncli" % ("-" * 1,), c="90")
|
||||
|
||||
spins = 0
|
||||
while self.ncli >= self.nclimax:
|
||||
if not spins:
|
||||
t = "at connection limit (global-option 'nc'); waiting"
|
||||
self.log(self.name, t, 3)
|
||||
|
||||
spins += 1
|
||||
time.sleep(0.1)
|
||||
if spins != 50 or not self.args.aclose:
|
||||
continue
|
||||
|
||||
ipfreq: dict[str, int] = {}
|
||||
with self.mutex:
|
||||
for c in self.clients:
|
||||
ip = ipnorm(c.ip)
|
||||
try:
|
||||
ipfreq[ip] += 1
|
||||
except:
|
||||
ipfreq[ip] = 1
|
||||
|
||||
ip, n = sorted(ipfreq.items(), key=lambda x: x[1], reverse=True)[0]
|
||||
if n < self.nclimax / 2:
|
||||
continue
|
||||
|
||||
self.aclose[ip] = int(time.time() + self.args.aclose * 60)
|
||||
nclose = 0
|
||||
nloris = 0
|
||||
nconn = 0
|
||||
with self.mutex:
|
||||
for c in self.clients:
|
||||
cip = ipnorm(c.ip)
|
||||
if ip != cip:
|
||||
continue
|
||||
|
||||
nconn += 1
|
||||
try:
|
||||
if (
|
||||
c.nreq >= 1
|
||||
or not c.cli
|
||||
or c.cli.in_hdr_recv
|
||||
or c.cli.keepalive
|
||||
):
|
||||
Daemon(c.shutdown)
|
||||
nclose += 1
|
||||
if c.nreq <= 0 and (not c.cli or c.cli.in_hdr_recv):
|
||||
nloris += 1
|
||||
except:
|
||||
pass
|
||||
|
||||
t = "{} downgraded to connection:close for {} min; dropped {}/{} connections"
|
||||
self.log(self.name, t.format(ip, self.args.aclose, nclose, nconn), 1)
|
||||
|
||||
if nloris < nconn / 2:
|
||||
continue
|
||||
|
||||
t = "slow%s (idle-conn): %s banned for %d min" # slowloris
|
||||
self.log(self.name, t % ("loris", ip, self.args.loris), 1)
|
||||
self.bans[ip] = int(time.time() + self.args.loris * 60)
|
||||
|
||||
if self.args.log_conn:
|
||||
self.log(self.name, "|%sC-acc1" % ("-" * 2,), c="90")
|
||||
|
||||
try:
|
||||
sck, saddr = srv_sck.accept()
|
||||
if tcp:
|
||||
cip = unicode(saddr[0])
|
||||
if cip.startswith("::ffff:"):
|
||||
cip = cip[7:]
|
||||
addr = (cip, saddr[1])
|
||||
else:
|
||||
addr = ("127.8.3.7", sck.fileno())
|
||||
except (OSError, socket.error) as ex:
|
||||
if self.stopping:
|
||||
break
|
||||
|
||||
self.log(self.name, "accept({}): {}".format(fno, ex), c=6)
|
||||
time.sleep(0.02)
|
||||
continue
|
||||
except TypeError as ex:
|
||||
# on macOS, accept() may return a None saddr if blocked by LittleSnitch;
|
||||
# unicode(saddr[0]) ==> TypeError: 'NoneType' object is not subscriptable
|
||||
if tcp and not saddr:
|
||||
t = "accept(%s): failed to accept connection from client due to firewall or network issue"
|
||||
self.log(self.name, t % (fno,), c=3)
|
||||
try:
|
||||
sck.close() # type: ignore
|
||||
except:
|
||||
pass
|
||||
time.sleep(0.02)
|
||||
continue
|
||||
raise
|
||||
|
||||
if self.args.log_conn:
|
||||
t = "|{}C-acc2 \033[0;36m{} \033[3{}m{}".format(
|
||||
"-" * 3, ip, port % 8, port
|
||||
)
|
||||
self.log("%s %s" % addr, t, c="90")
|
||||
|
||||
self.accept(sck, addr)
|
||||
|
||||
def accept(self, sck: socket.socket, addr: tuple[str, int]) -> None:
|
||||
"""takes an incoming tcp connection and creates a thread to handle it"""
|
||||
now = time.time()
|
||||
|
||||
if now - (self.tp_time or now) > 300:
|
||||
t = "httpserver threadpool died: tpt {:.2f}, now {:.2f}, nthr {}, ncli {}"
|
||||
self.log(self.name, t.format(self.tp_time, now, self.tp_nthr, self.ncli), 1)
|
||||
self.tp_time = 0
|
||||
self.tp_q = None
|
||||
|
||||
with self.mutex:
|
||||
self.ncli += 1
|
||||
if not self.t_periodic:
|
||||
name = "hsrv-pt"
|
||||
if self.nid:
|
||||
name += "-%d" % (self.nid,)
|
||||
|
||||
self.t_periodic = Daemon(self.periodic, name)
|
||||
|
||||
if self.tp_q:
|
||||
self.tp_time = self.tp_time or now
|
||||
self.tp_ncli = max(self.tp_ncli, self.ncli)
|
||||
if self.tp_nthr < self.ncli + 4:
|
||||
self.start_threads(8)
|
||||
|
||||
self.tp_q.put((sck, addr))
|
||||
return
|
||||
|
||||
if not self.args.no_htp:
|
||||
t = "looks like the httpserver threadpool died; please make an issue on github and tell me the story of how you pulled that off, thanks and dog bless\n"
|
||||
self.log(self.name, t, 1)
|
||||
|
||||
Daemon(
|
||||
self.thr_client,
|
||||
"httpconn-%s-%d" % (addr[0].split(".", 2)[-1][-6:], addr[1]),
|
||||
(sck, addr),
|
||||
)
|
||||
|
||||
def thr_poolw(self) -> None:
|
||||
assert self.tp_q # !rm
|
||||
while True:
|
||||
task = self.tp_q.get()
|
||||
if not task:
|
||||
break
|
||||
|
||||
with self.mutex:
|
||||
self.tp_time = 0
|
||||
|
||||
try:
|
||||
sck, addr = task
|
||||
me = threading.current_thread()
|
||||
me.name = "httpconn-%s-%d" % (addr[0].split(".", 2)[-1][-6:], addr[1])
|
||||
self.thr_client(sck, addr)
|
||||
me.name = self.name + "-poolw"
|
||||
except Exception as ex:
|
||||
if str(ex).startswith("client d/c "):
|
||||
self.log(self.name, "thr_client: " + str(ex), 6)
|
||||
else:
|
||||
self.log(self.name, "thr_client: " + min_ex(), 3)
|
||||
|
||||
def shutdown(self) -> None:
|
||||
self.stopping = True
|
||||
for srv in self.srvs:
|
||||
try:
|
||||
srv.close()
|
||||
except:
|
||||
pass
|
||||
|
||||
thrs = []
|
||||
clients = list(self.clients)
|
||||
for cli in clients:
|
||||
t = threading.Thread(target=cli.shutdown)
|
||||
thrs.append(t)
|
||||
t.start()
|
||||
|
||||
if self.tp_q:
|
||||
self.stop_threads(self.tp_nthr)
|
||||
for _ in range(10):
|
||||
time.sleep(0.05)
|
||||
if self.tp_q.empty():
|
||||
break
|
||||
|
||||
for t in thrs:
|
||||
t.join()
|
||||
|
||||
self.log(self.name, "ok bye")
|
||||
|
||||
def thr_client(self, sck: socket.socket, addr: tuple[str, int]) -> None:
|
||||
"""thread managing one tcp client"""
|
||||
cli = HttpConn(sck, addr, self)
|
||||
with self.mutex:
|
||||
self.clients.add(cli)
|
||||
|
||||
# print("{}\n".format(len(self.clients)), end="")
|
||||
fno = sck.fileno()
|
||||
try:
|
||||
if self.args.log_conn:
|
||||
self.log("%s %s" % addr, "|%sC-crun" % ("-" * 4,), c="90")
|
||||
|
||||
cli.run()
|
||||
|
||||
except (OSError, socket.error) as ex:
|
||||
if ex.errno not in E_SCK:
|
||||
self.log(
|
||||
"%s %s" % addr,
|
||||
"run({}): {}".format(fno, ex),
|
||||
c=6,
|
||||
)
|
||||
|
||||
finally:
|
||||
sck = cli.s
|
||||
if self.args.log_conn:
|
||||
self.log("%s %s" % addr, "|%sC-cdone" % ("-" * 5,), c="90")
|
||||
|
||||
try:
|
||||
fno = sck.fileno()
|
||||
shut_socket(cli.log, sck)
|
||||
except (OSError, socket.error) as ex:
|
||||
if not MACOS:
|
||||
self.log(
|
||||
"%s %s" % addr,
|
||||
"shut({}): {}".format(fno, ex),
|
||||
c="90",
|
||||
)
|
||||
if ex.errno not in E_SCK:
|
||||
raise
|
||||
finally:
|
||||
with self.mutex:
|
||||
self.clients.remove(cli)
|
||||
self.ncli -= 1
|
||||
|
||||
if cli.u2idx:
|
||||
self.put_u2idx(str(addr), cli.u2idx)
|
||||
|
||||
def cachebuster(self) -> str:
|
||||
if time.time() - self.cb_ts < 1:
|
||||
return self.cb_v
|
||||
|
||||
with self.mutex:
|
||||
if time.time() - self.cb_ts < 1:
|
||||
return self.cb_v
|
||||
|
||||
v = self.E.t0
|
||||
try:
|
||||
with os.scandir(self.E.mod_ + "web") as dh:
|
||||
for fh in dh:
|
||||
inf = fh.stat()
|
||||
v = max(v, inf.st_mtime)
|
||||
except:
|
||||
pass
|
||||
|
||||
# spack gives 4 lsb, take 3 lsb, get 4 ch
|
||||
self.cb_v = ub64enc(spack(b">L", int(v))[1:]).decode("ascii")
|
||||
self.cb_ts = time.time()
|
||||
return self.cb_v
|
||||
|
||||
def get_u2idx(self, ident: str) -> Optional[U2idx]:
|
||||
utab = self.u2idx_free
|
||||
for _ in range(100): # 5/0.05 = 5sec
|
||||
with self.mutex:
|
||||
if utab:
|
||||
if ident in utab:
|
||||
return utab.pop(ident)
|
||||
|
||||
return utab.pop(list(utab.keys())[0])
|
||||
|
||||
if self.u2idx_n < CORES:
|
||||
self.u2idx_n += 1
|
||||
return U2idx(self)
|
||||
|
||||
time.sleep(0.05)
|
||||
# not using conditional waits, on a hunch that
|
||||
# average performance will be faster like this
|
||||
# since most servers won't be fully saturated
|
||||
|
||||
return None
|
||||
|
||||
def put_u2idx(self, ident: str, u2idx: U2idx) -> None:
|
||||
with self.mutex:
|
||||
while ident in self.u2idx_free:
|
||||
ident += "a"
|
||||
|
||||
self.u2idx_free[ident] = u2idx
|
||||
|
||||
def read_dls(
|
||||
self,
|
||||
) -> tuple[
|
||||
dict[str, tuple[float, int, str, str, str]], dict[str, tuple[float, int]]
|
||||
]:
|
||||
"""
|
||||
mp-broker asking for local dl-info + dl-state;
|
||||
reduce overhead by sending just the vfs vpath
|
||||
"""
|
||||
dli = {k: (a, b, c.vpath, d, e) for k, (a, b, c, d, e) in self.dli.items()}
|
||||
return (dli, self.dls)
|
||||
|
||||
def write_dls(
|
||||
self,
|
||||
sdli: dict[str, tuple[float, int, str, str, str]],
|
||||
dls: dict[str, tuple[float, int]],
|
||||
) -> None:
|
||||
"""
|
||||
mp-broker pushing total dl-info + dl-state;
|
||||
swap out the vfs vpath with the vfs node
|
||||
"""
|
||||
dli: dict[str, tuple[float, int, "VFS", str, str]] = {}
|
||||
for k, (a, b, c, d, e) in sdli.items():
|
||||
vn = self.asrv.vfs.all_nodes[c]
|
||||
dli[k] = (a, b, vn, d, e)
|
||||
|
||||
self.tdli = dli
|
||||
self.tdls = dls
|
||||
|
||||
def pregen_thumbs(self) -> None:
|
||||
Daemon(self._pregen_thumbs, "th_pregen")
|
||||
|
||||
def _pregen_thumbs(self) -> None:
|
||||
def log(msg, n):
|
||||
self.log("thumb-pregen", msg, n)
|
||||
|
||||
if getattr(self, "_pregen", False):
|
||||
log("already running", 1)
|
||||
return
|
||||
|
||||
self._pregen = True
|
||||
|
||||
for n in range(9999999):
|
||||
x = self.broker.ask("up2k.is_busy")
|
||||
zb, zi = x.get()
|
||||
if zi:
|
||||
break
|
||||
if not n:
|
||||
log("waiting for up2k to finish initializing", 6)
|
||||
time.sleep(1 if n < 10 else 5 if n < 300 else 15)
|
||||
|
||||
if not self.thumbcli:
|
||||
log("no thumbcli", 1)
|
||||
return
|
||||
|
||||
if self.args.th_pre_rl:
|
||||
try:
|
||||
self.broker.hub.thumbsrv.log = self.broker.hub.thumbsrv._slog
|
||||
except:
|
||||
pass
|
||||
|
||||
nfiles = 0
|
||||
t0 = time.time()
|
||||
scandir = not self.args.no_scandir
|
||||
for vn in self.asrv.vfs.all_nodes.values():
|
||||
fmts = vn.flags.get("th_pregen", "")
|
||||
if not fmts:
|
||||
continue
|
||||
log("starting for volume /%s" % (vn.vpath,), 6)
|
||||
g = vn.walk("", "/", [], LEELOO_DALLAS, [[True]], 2, scandir, False, False)
|
||||
g = gfilter2(g, self, vn.vpath, fmts.split(","))
|
||||
for f in g:
|
||||
nfiles += 1
|
||||
if not nfiles % 256:
|
||||
now = time.time()
|
||||
for n in range(9999999):
|
||||
x = self.broker.ask("up2k.is_busy")
|
||||
zb, zi = x.get()
|
||||
if not zb:
|
||||
if n:
|
||||
t0 += time.time() - now
|
||||
break
|
||||
if not n:
|
||||
log("waiting for up2k to finish indexing", 6)
|
||||
time.sleep(5)
|
||||
|
||||
if self.args.th_pre_rl:
|
||||
try:
|
||||
self.broker.hub.thumbsrv.log = self.broker.hub.thumbsrv._log
|
||||
except:
|
||||
pass
|
||||
|
||||
t = "finished; %d files in %d seconds"
|
||||
log(t % (nfiles, time.time() - t0), 6)
|
||||
self._pregen = False
|
||||
|
||||
if self.args.exit == "thgen":
|
||||
self.broker.say("sigterm")
|
||||
Reference in New Issue
Block a user