diff --git a/STUDIO.md b/STUDIO.md
index 9176e98..dbfa0a6 100644
--- a/STUDIO.md
+++ b/STUDIO.md
@@ -201,3 +201,9 @@ Die Update-Abfrage zeigt das aktuelle reguläre Release (`releases/latest`) zuer
Unter GPU-Auswahl: Automatisch (maximaler Wert aus 512 MiB und 2,5 Prozent je GPU), Manuell (0–32768 MiB je ausgewählter GPU) oder Keine zusätzliche Reserve. Manuelle Werte werden per GPU-UUID gespeichert. Alte Profile verwenden weiter Automatisch. Die Reserve beeinflusst Fit-Ziel und Speicherentscheidung; Modell/KV/MTP sowie die konservative Projektor-Schätzung (Dateigröße plus 2 GiB Arbeitsbedarf) bleiben enthalten. Host-RAM-Grenzen bleiben bestehen. Ohne Reserve ist ein erfolgreicher Modellstart nicht garantiert.
Für einen Versuch ohne CPU-Offloading zusätzlich „Vollständig auf GPU · sonst Start abbrechen“ wählen. Dann gibt es bei negativer Prognose keinen automatischen Layer-Rückfall. „Automatisch“ unter GPU-Ausführung erlaubt ihn weiterhin ausdrücklich. Der Testchat zeigt Layerlimit und Reserve je GPU. Gespeicherte Profile werden durch dieses Update nicht umgestellt.
+
+### Download-Warteschlange und Geschwindigkeit
+
+Während eines Downloads können bis zu 20 weitere Dateien eingereiht werden, auch kategorienübergreifend. Genau ein Transfer läuft; nach Abschluss, Fehler oder manuellem Abbruch startet der nächste. Doppelte Dateien werden abgewiesen. Unter Downloads lassen sich wartende Einträge entfernen. Die Speicherprüfung berücksichtigt noch ausstehende Bytes aller geplanten Transfers plus 10 GiB Reserve.
+
+Die API liefert `queued`, `bytes_per_second` und `eta_seconds`. Geschwindigkeit basiert auf einem gleitenden Zeitfenster; ETA ist eine Schätzung und enthält keine abschließende Dateiverarbeitung. Vor ausreichenden Messwerten bleibt die Anzeige offen. Beim Serverneustart werden laufende und wartende Einträge als unterbrochen markiert; keine automatische Wiederaufnahme.
diff --git a/catalog-ui.js b/catalog-ui.js
index e0534b2..e8f0814 100644
--- a/catalog-ui.js
+++ b/catalog-ui.js
@@ -32,8 +32,8 @@ window.CatalogUI = (() => {
el('hub-capacity').textContent = `Athena · ${size(state.free_bytes)} freier Speicher · 10 GiB bleiben als Reserve frei`;
const jobs=downloads ? (state.downloads||[]).filter(j=>j.kind===kind) : state.job?.state==='downloading' ? [state.job] : [];
const signature=JSON.stringify(jobs); if(signature===lastJobs)return; lastJobs=signature;
- const labels={downloading:'Wird heruntergeladen',complete:'Download abgeschlossen',cancelled:'Download abgebrochen',failed:'Download fehlgeschlagen',interrupted:'Download unterbrochen'};
- el('hub-job').innerHTML=jobs.map(j=>{const percent=j.total>0?Math.min(100,Math.max(0,j.bytes/j.total*100)):0;return `
`:''}`:'Wähle eine Datei für die Speicherprüfung. Gesamtbedarf der Laufzeit noch nicht berechnet.';
}
@@ -83,7 +84,7 @@ window.CatalogUI = (() => {
const chosen = el('hub-variants').querySelector('input:checked'); if (!chosen || starting) return;
const request = {repo:detail.repo,revision:detail.revision,filename:detail.files[Number(chosen.value)].name,kind};
starting = true; updateSelection();
- try { await api('/download', request); message('Download gestartet. Den Fortschritt siehst du oberhalb der Galerie.'); await refresh(); }
+ try { const job=await api('/download', request); message(job.state==='queued'?'In Warteschlange eingereiht. Übersicht unter Downloads.':'Download gestartet. Fortschritt unter Downloads.'); await refresh(); }
catch(error) { message(error.message); }
finally { starting = false; if (root.isConnected) updateSelection(); }
};
diff --git a/catalog.py b/catalog.py
index 9cee36e..7e2d923 100644
--- a/catalog.py
+++ b/catalog.py
@@ -70,12 +70,12 @@ def capabilities(model):
class Catalog:
def __init__(self,root):
self.root=Path(root);self.lock=threading.RLock();self.job=None;self.cancel=threading.Event()
- self.cache={};self.cache_lock=threading.Lock();self.history=[]
+ self.cache={};self.cache_lock=threading.Lock();self.history=[];self.pending=[];self.closing=False;self.samples=[]
path=self.root/'downloads.json'
if path.exists():
self.history=json.loads(path.read_text())
for job in self.history:
- if job['state']=='downloading':job.update(state='interrupted',error='Deck wurde neu gestartet. Datei erneut auswählen und herunterladen.')
+ if job['state'] in ('downloading','queued'):job.update(state='interrupted',error='Deck wurde neu gestartet. Datei erneut auswählen und herunterladen.')
else:
for entry in self.root.glob('*/entry.json'):
try:
@@ -91,6 +91,9 @@ class Catalog:
job=next((x for x in self.history if x['id']==job_id),None)
if not job:raise ValueError('Download nicht gefunden.')
if job['state']=='downloading':raise ValueError('Laufenden Download zuerst abbrechen.')
+ if job['state']=='queued':
+ self.pending=[x for x in self.pending if x[0]['id']!=job_id]
+ job['state']='cancelled'
job['dismissed']=True
if self.job and self.job.get('id')==job_id:self.job=None
self._save_history()
@@ -161,7 +164,8 @@ class Catalog:
def start(self,repo,filename,revision,kind):
if kind not in KINDS:raise ValueError('Ungültiger Bereich.')
with self.lock:
- if self.job and self.job['state']=='downloading':raise ValueError('Ein Download läuft bereits.')
+ if self.closing:raise ValueError('Deck wird beendet.')
+ if len(self.pending)>=20:raise ValueError('Warteschlange voll (20 Dateien).')
with self.cache_lock:self.cache.pop(repo,None)
data=self.files(repo)
if data['gated']:raise ValueError('Zugangsbeschränkte Modelle werden noch nicht unterstützt.')
@@ -169,18 +173,25 @@ class Catalog:
item=next((x for x in data['files'] if x['name']==filename),None)
if not item:raise ValueError('Datei nicht verfügbar.')
self.root.mkdir(parents=True,exist_ok=True,mode=0o700)
- if item['size']>shutil.disk_usage(self.root).free-10*1024**3:raise ValueError('Nicht genug Platz mit 10 GiB freier Reserve.')
+ reserved=sum(x['total']-x['bytes'] for x in self.history if x['state'] in ('downloading','queued'))
+ if item['size']+reserved>shutil.disk_usage(self.root).free-10*1024**3:raise ValueError('Nicht genug Platz mit 10 GiB freier Reserve.')
ident=hashlib.sha256((repo+revision+filename).encode()).hexdigest()
+ if any(x.get('entry_id')==ident and x['state'] in ('downloading','queued') for x in self.history):raise ValueError('Datei läuft bereits oder steht in der Warteschlange.')
target=self.root/ident
if (target/'entry.json').exists():raise ValueError('Datei bereits in der Bibliothek.')
target.mkdir(exist_ok=True,mode=0o700)
- self.cancel.clear()
- self.job=dict(id=uuid.uuid4().hex,entry_id=ident,state='downloading',repo=repo,file=filename,kind=kind,revision=revision,created_at=time.time(),bytes=0,total=item['size'],error=None)
- self.history.append(self.job);self._save_history()
- threading.Thread(target=self._download,args=(data,item,target,kind),daemon=True).start()
- return dict(self.job)
- def stop(self):
- self.cancel.set()
+ job=dict(id=uuid.uuid4().hex,entry_id=ident,state='queued',repo=repo,file=filename,kind=kind,revision=revision,created_at=time.time(),bytes=0,total=item['size'],error=None,bytes_per_second=None,eta_seconds=None)
+ self.history.append(job);self.pending.append((job,data,item,target,kind));self._next();self._save_history()
+ return dict(job)
+ def _next(self):
+ if self.closing or (self.job and self.job['state']=='downloading') or not self.pending:return
+ self.job,data,item,target,kind=self.pending.pop(0)
+ self.job['state']='downloading';self.cancel.clear();self.samples=[(time.monotonic(),0)]
+ threading.Thread(target=self._download,args=(data,item,target,kind),daemon=True).start()
+ def stop(self,shutdown=False):
+ with self.lock:
+ if shutdown:self.closing=True
+ self.cancel.set()
return {'cancellation_requested':True}
def _download(self,data,item,target,kind):
partial=target/'download.part';dest=target/('model'+PurePosixPath(item['name']).suffix)
@@ -195,17 +206,22 @@ class Catalog:
received+=len(chunk)
if received>item['size'] or shutil.disk_usage(target).free<10*1024**3:raise ValueError('Größe oder Speicherreserve überschritten.')
out.write(chunk);digest.update(chunk)
- with self.lock:self.job['bytes']=received
+ with self.lock:
+ now=time.monotonic();self.samples.append((now,received))
+ while len(self.samples)>2 and self.samples[1][0]=.25 else None
+ self.job.update(bytes=received,bytes_per_second=speed,eta_seconds=(item['size']-received)/speed if speed else None)
if received!=item['size']:raise ValueError('Unvollständiger Download.')
if item['sha256'] and digest.hexdigest()!=item['sha256']:raise ValueError('SHA-256-Prüfung fehlgeschlagen.')
partial.replace(dest)
entry=dict(repo=data['repo'],revision=data['revision'],file=item['name'],size=received,sha256=digest.hexdigest(),upstream_hash_verified=bool(item['sha256']),kind=kind,downloaded_at=time.time(),state='downloaded',runtime_ready=False)
temp=target/'entry.tmp';temp.write_text(json.dumps(entry));temp.replace(target/'entry.json')
- with self.lock:self.job['state']='complete'
+ terminal='complete';error=None
except Exception as exc:
- with self.lock:
- self.job['state']='cancelled' if isinstance(exc,InterruptedError) else 'failed'
- self.job['error']=str(exc) if isinstance(exc,ValueError) else ('Abgebrochen.' if isinstance(exc,InterruptedError) else 'Download fehlgeschlagen; Verbindung oder Anbieter prüfen.')
+ terminal='cancelled' if isinstance(exc,InterruptedError) else 'failed'
+ error=str(exc) if isinstance(exc,ValueError) else ('Abgebrochen.' if isinstance(exc,InterruptedError) else 'Download fehlgeschlagen; Verbindung oder Anbieter prüfen.')
partial.unlink(missing_ok=True)
finally:
- with self.lock:self._save_history()
+ with self.lock:
+ self.job.update(state=terminal,error=error,bytes_per_second=None,eta_seconds=None)
+ self._save_history();self._next()
diff --git a/server.py b/server.py
index 949274d..7f00541 100644
--- a/server.py
+++ b/server.py
@@ -450,7 +450,7 @@ def main():
server.image_runtime.stop()
server.image_tests.stop()
server.runtime.stop()
- server.catalog.stop()
+ server.catalog.stop(shutdown=True)
server.server_close()
if __name__ == '__main__':
diff --git a/test_catalog.py b/test_catalog.py
index ea71ace..27e1f46 100644
--- a/test_catalog.py
+++ b/test_catalog.py
@@ -66,6 +66,31 @@ class CatalogTests(unittest.TestCase):
with self.assertRaises(ValueError):c.start('a/b','m.gguf','a'*40,'chat')
with patch('catalog.shutil.disk_usage',return_value=type('Disk',(),{'free':1024})()):
with self.assertRaises(ValueError):c.start('a/b','m.gguf','b'*40,'chat')
+ def test_queue_progress_duplicate_and_removal(self):
+ import threading
+ release=threading.Event();progress=threading.Event()
+ class Stream(io.BytesIO):
+ def read(self,n=-1):
+ if self.tell()==0:
+ time.sleep(.3);return super().read(1)
+ progress.set();release.wait(3);return super().read(n)
+ with tempfile.TemporaryDirectory() as d:
+ c=Catalog(d);data=dict(repo='a/b',revision='a'*40,gated=False,files=[dict(name=n,size=2,sha256=None) for n in ['a.gguf','b.gguf','c.gguf']])
+ with patch.object(c,'files',return_value=data),patch('catalog.remote',side_effect=[Stream(b'aa'),io.BytesIO(b'bb')]),patch('catalog.shutil.disk_usage',return_value=type('Disk',(),{'free':100*1024**3})()):
+ first=c.start('a/b','a.gguf','a'*40,'chat');self.assertTrue(progress.wait(2))
+ status=c.status()['job'];self.assertGreater(status['bytes_per_second'],0);self.assertGreater(status['eta_seconds'],0)
+ second=c.start('a/b','b.gguf','a'*40,'chat');self.assertEqual(second['state'],'queued')
+ with self.assertRaises(ValueError):c.start('a/b','b.gguf','a'*40,'chat')
+ third=c.start('a/b','c.gguf','a'*40,'chat');c.dismiss(third['id'])
+ release.set()
+ for _ in range(200):
+ if len(c.status()['entries'])==2 and c.status()['job']['state']=='complete':break
+ time.sleep(.01)
+ self.assertEqual(len(c.status()['entries']),2);self.assertEqual(c.status()['job']['id'],second['id']);self.assertFalse(c.pending)
+ def test_shutdown_does_not_start_queued_file(self):
+ with tempfile.TemporaryDirectory() as d:
+ c=Catalog(d);c.pending=[('sentinel',)];c.stop(shutdown=True)
+ with patch('catalog.threading.Thread') as thread:c._next();thread.assert_not_called()
def test_cancellation(self):
with tempfile.TemporaryDirectory() as directory:
c=Catalog(directory);c.job=dict(state='downloading');c.stop()