ston1th/cachefs
0
0
Fork 0
Code Issues Releases Activity

added back anchor and selfheal broken chunks

Browse source
This commit is contained in:
ston1th 2022-10-09 01:12:05 +02:00
commit 6122768619
8 changed files with 65 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

8
cmd/cachefs/main.go
View file

@ -64,8 +64,8 @@ func main() {
slog,
),
}
log.Info("starting main server", "addr", c.Server.HTTP.Addr)
go func() {
log.Info("starting main server", "addr", c.Server.HTTP.Addr)
err := s.ListenAndServe()
if err != nil && err != http.ErrServerClosed {
klog.Fatalf("init failed: %s", err)
@ -78,7 +78,6 @@ func main() {
)
davAddr := c.Server.WebDav.Addr
if davAddr != "" {
log.Info("starting webdav server", "addr", davAddr)
dav = &http.Server{
Addr: davAddr,
Handler: srv.NewWebDavServer(
@ -86,6 +85,7 @@ func main() {
slog.WithName("webdav"),
),
}
log.Info("starting webdav server", "addr", davAddr)
go func() {
err := dav.ListenAndServe()
if err != nil && err != http.ErrServerClosed {
@ -95,7 +95,6 @@ func main() {
}
cacheAddr := c.Server.Cache.Addr
if cacheAddr != "" {
log.Info("starting cache server", "addr", cacheAddr)
cache = &http.Server{
Addr: cacheAddr,
Handler: srv.NewCacheServer(
@ -103,6 +102,7 @@ func main() {
slog.WithName("cache"),
),
}
log.Info("starting cache server", "addr", cacheAddr)
go func() {
err := cache.ListenAndServe()
if err != nil && err != http.ErrServerClosed {
@ -112,7 +112,6 @@ func main() {
}
plainAddr := c.Server.Plain.Addr
if plainAddr != "" {
log.Info("starting plain server", "addr", plainAddr)
plain = &http.Server{
Addr: plainAddr,
Handler: srv.NewPlainServer(
@ -120,6 +119,7 @@ func main() {
slog.WithName("plain"),
),
}
log.Info("starting plain server", "addr", plainAddr)
go func() {
err := plain.ListenAndServe()
if err != nil && err != http.ErrServerClosed {

6
pkg/fs/file.go
View file

@ -98,8 +98,12 @@ func (f *File) Read(p []byte) (n int, err error) {
if err == io.EOF {
return
}
if err == crypto.ErrDecrypt {
log.V(2).Info("rereading to cache file", "err", err)
return f.readToCache(p)
}
if err != nil {
if !IsIOErr(err) && err != crypto.ErrDecrypt {
if !IsIOErr(err) {
log.Error(err, "error reading cache file")
return
}

8
pkg/fs/fs.go
View file

@ -129,13 +129,13 @@ const (
func FileSize(s int64) string {
switch {
case s >= TiB:
return strconv.Itoa(int(s/TiB)) + " TiB"
return strconv.FormatFloat(float64(s)/TiB, 'f', 2, 64) + " TiB"
case s >= GiB:
return strconv.Itoa(int(s/GiB)) + " GiB"
return strconv.FormatFloat(float64(s)/GiB, 'f', 2, 64) + " GiB"
case s >= MiB:
return strconv.Itoa(int(s/MiB)) + " MiB"
return strconv.FormatFloat(float64(s)/MiB, 'f', 2, 64) + " MiB"
case s >= KiB:
return strconv.Itoa(int(s/KiB)) + " KiB"
return strconv.FormatFloat(float64(s)/KiB, 'f', 2, 64) + " KiB"
}
return strconv.Itoa(int(s)) + " B"
}

30
pkg/fs/metadata.go
View file

@ -381,6 +381,7 @@ func (md *Metadata) close() (err error) {
md.cancel()
if md.wc != nil {
close(md.wc)
<-md.done
md.wc = nil
}
if md.f != nil {
@ -449,30 +450,27 @@ type writeAt struct {
func (md *Metadata) resetChans() {
md.wc = make(chan writeAt, 10)
md.done = make(chan struct{})
}
func (md *Metadata) writer() {
md.resetChans()
md.err.Store(nil)
go func() {
for {
select {
case <-md.ctx.Done():
return
case wa := <-md.wc:
atomic.StoreInt64(&md.Atime, now())
n, err := md.f.WriteAt(wa.data, wa.pos)
if err == nil {
md.addChunk(wa.pos, n)
md.fs.q.Add(n)
} else {
md.err.Store(&mdErr{err})
}
if wa.ret != nil {
streamingPool.Put(wa.ret)
}
for wa := range md.wc {
atomic.StoreInt64(&md.Atime, now())
n, err := md.f.WriteAt(wa.data, wa.pos)
if err == nil {
md.addChunk(wa.pos, n)
md.fs.q.Add(n)
} else {
md.err.Store(&mdErr{err})
}
if wa.ret != nil {
streamingPool.Put(wa.ret)
}
}
close(md.done)
}()
}

4
pkg/srv/cache.go
View file

@ -82,7 +82,7 @@ func (cs *CacheServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
h.Del("Last-Modified")
paths, err := i.GetPaths(p, cs.fs, true)
paths, pathAnchor, err := i.GetPaths(p, cs.fs, true)
if err == io.EOF {
w.WriteHeader(i.Status())
return
@ -94,7 +94,7 @@ func (cs *CacheServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
h.Set(csp, indexCSP)
w.WriteHeader(i.Status())
err = cache.Execute(w, data{Paths: paths})
err = cache.Execute(w, data{Paths: paths, PathAnchor: pathAnchor})
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return

38
pkg/srv/interceptor.go
View file

@ -47,8 +47,9 @@ type dirContents struct {
}
type dir struct {
Name template.HTML
URI template.HTML
Name template.HTML
URI template.HTML
Anchor string
}
type dirs []dir
@ -96,7 +97,19 @@ func (r *responseInterceptor) Status() int {
return r.status
}
func (r *responseInterceptor) GetPaths(path string, filesystem *fs.FS, relative bool) (dc dirContents, err error) {
func xmldecode(s string) string {
s = strings.ReplaceAll(s, "&amp;", "&")
s = strings.ReplaceAll(s, "&#39;", "'")
return s
}
func urlencode(s string) string {
s = strings.ReplaceAll(s, "[", "%5B")
s = strings.ReplaceAll(s, "]", "%5D")
return s
}
func (r *responseInterceptor) GetPaths(path string, filesystem *fs.FS, relative bool) (dc dirContents, pa string, err error) {
buf := r.buf.Bytes()
buf = bytes.ReplaceAll(buf, []byte{'&'}, []byte("&#38;"))
err = xml.Unmarshal(buf, &dc)
@ -105,19 +118,20 @@ func (r *responseInterceptor) GetPaths(path string, filesystem *fs.FS, relative
}
dc.Base = path
path = strings.TrimSuffix(path, "/")
pa = "#" + anchor(path)
for _, p := range dc.AllPaths {
p = strings.ReplaceAll(p, "&amp;", "&")
p = strings.ReplaceAll(p, "&#39;", "'")
p = xmldecode(p)
name := template.HTML(p)
full := path + "/" + p
uri := template.HTML(full)
uri := template.HTML(urlencode(full))
if relative {
uri = template.HTML(p)
uri = template.HTML(urlencode(p))
}
if p[len(p)-1] == '/' {
dc.Dirs = append(dc.Dirs, dir{
Name: name,
URI: uri,
Name: name,
URI: uri,
Anchor: pathAnchor(p),
})
} else {
dc.Files = append(dc.Files, file{
@ -139,7 +153,7 @@ func getPreloads(path string, filesystem *fs.FS) (dc dirContents) {
for _, p := range filesystem.Preloads() {
dc.Files = append(dc.Files, file{
Name: template.HTML(p.Name),
URI: template.HTML(p.Name),
URI: template.HTML(urlencode(p.Name)),
Size: fs.FileSize(p.Size),
Rate: p.Rate(),
Status: p.Status,
@ -151,6 +165,10 @@ func getPreloads(path string, filesystem *fs.FS) (dc dirContents) {
return
}
func pathAnchor(s string) string {
return anchor(strings.TrimSuffix(s, "/"))
}
func anchor(s string) string {
if i := strings.LastIndex(s, "/"); i > 0 && i+1 < len(s) {
s = s[i+1:]

11
pkg/srv/srv.go
View file

@ -25,9 +25,10 @@ const (
)
type data struct {
QuotaCur float64
QuotaMax float64
Paths dirContents
QuotaCur float64
QuotaMax float64
Paths dirContents
PathAnchor string
}
type FileServer struct {
@ -136,7 +137,7 @@ func (fs *FileServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
h.Del("Last-Modified")
paths, err := i.GetPaths(p, fs.fs, false)
paths, pathAnchor, err := i.GetPaths(p, fs.fs, false)
if err == io.EOF {
w.WriteHeader(i.Status())
return
@ -148,7 +149,7 @@ func (fs *FileServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
h.Set(csp, indexCSP)
w.WriteHeader(i.Status())
err = index.Execute(w, data{Paths: paths})
err = index.Execute(w, data{Paths: paths, PathAnchor: pathAnchor})
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return

4
pkg/srv/templates/index.html
View file

@ -5,13 +5,13 @@
<a href="/?o=preloads">[Preloads]</a></pre>
<table>
<tr class="listitem">
<td class="listpath up"><a href="../">../</a></td>
<td class="listpath up"><a href="../{{.PathAnchor}}">../</a></td>
<td class="listoptions">[dir]</td>
</tr>
<tr class="spacer"><td colspan="2"></td></tr>
{{range $s := .Paths.Dirs -}}
<tr class="listitem">
<td class="listpath"><a href="{{$s.URI}}">{{$s.Name}}</a></td>
<td class="listpath"><a id="{{$s.Anchor}}" href="{{$s.URI}}">{{$s.Name}}</a></td>
<td class="listoptions">[dir]</td>
</tr>
<tr class="spacer"><td colspan="2"></td></tr>