mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-01-19 02:48:24 +00:00
enhance deletion operation
This commit is contained in:
parent
7c5c94785c
commit
7251e357e7
|
@ -1,8 +1,7 @@
|
|||
package operation
|
||||
|
||||
import (
|
||||
"code.google.com/p/weed-fs/go/glog"
|
||||
"net/http"
|
||||
"code.google.com/p/weed-fs/go/util"
|
||||
)
|
||||
|
||||
func DeleteFile(server string, fileId string) error {
|
||||
|
@ -10,14 +9,5 @@ func DeleteFile(server string, fileId string) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return Delete(fileUrl)
|
||||
}
|
||||
func Delete(url string) error {
|
||||
req, err := http.NewRequest("DELETE", url, nil)
|
||||
if err != nil {
|
||||
glog.V(0).Infoln("failing to delete", url)
|
||||
return err
|
||||
}
|
||||
_, err = client.Do(req)
|
||||
return err
|
||||
return util.Delete(fileUrl)
|
||||
}
|
||||
|
|
|
@ -5,6 +5,7 @@ import (
|
|||
"code.google.com/p/weed-fs/go/glog"
|
||||
"code.google.com/p/weed-fs/go/operation"
|
||||
"code.google.com/p/weed-fs/go/storage"
|
||||
"code.google.com/p/weed-fs/go/util"
|
||||
"net/http"
|
||||
"strconv"
|
||||
)
|
||||
|
@ -39,7 +40,7 @@ func ReplicatedWrite(masterNode string, s *storage.Store, volumeId storage.Volum
|
|||
volumeId.String() + ": " + err.Error()
|
||||
} else {
|
||||
distributedOperation(masterNode, s, volumeId, func(location operation.Location) bool {
|
||||
return nil == operation.Delete("http://"+location.Url+r.URL.Path+"?type=replicate")
|
||||
return nil == util.Delete("http://"+location.Url+r.URL.Path+"?type=replicate")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
@ -61,7 +62,7 @@ func ReplicatedDelete(masterNode string, store *storage.Store, volumeId storage.
|
|||
if needToReplicate { //send to other replica locations
|
||||
if r.FormValue("type") != "replicate" {
|
||||
if !distributedOperation(masterNode, store, volumeId, func(location operation.Location) bool {
|
||||
return nil == operation.Delete("http://"+location.Url+r.URL.Path+"?type=replicate")
|
||||
return nil == util.Delete("http://"+location.Url+r.URL.Path+"?type=replicate")
|
||||
}) {
|
||||
ret = 0
|
||||
}
|
||||
|
|
|
@ -52,3 +52,22 @@ func Get(url string) ([]byte, error) {
|
|||
}
|
||||
return b, nil
|
||||
}
|
||||
|
||||
func Delete(url string) error {
|
||||
req, err := http.NewRequest("DELETE", url, nil)
|
||||
if err != nil {
|
||||
glog.V(0).Infoln("failing to delete", url)
|
||||
return err
|
||||
}
|
||||
resp, e := client.Do(req)
|
||||
if e != nil {
|
||||
glog.V(0).Infoln(e)
|
||||
return e
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if _, err := ioutil.ReadAll(resp.Body); err != nil {
|
||||
glog.V(0).Infoln("read get result from", url, err)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
|
@ -166,6 +166,28 @@ func bench_read() {
|
|||
}
|
||||
|
||||
func writeFiles(idChan chan int, fileIdLineChan chan string, s *stats) {
|
||||
deleteChan := make(chan *operation.FilePart, 100)
|
||||
var waitForDeletions sync.WaitGroup
|
||||
time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
|
||||
for i := 0; i < 7; i++ {
|
||||
go func() {
|
||||
waitForDeletions.Add(1)
|
||||
for fp := range deleteChan {
|
||||
if fp == nil {
|
||||
break
|
||||
}
|
||||
serverLimitChan[fp.Server] <- true
|
||||
if e := util.Delete("http://" + fp.Server + "/" + fp.Fid); e == nil {
|
||||
s.completed++
|
||||
} else {
|
||||
s.failed++
|
||||
}
|
||||
<-serverLimitChan[fp.Server]
|
||||
}
|
||||
waitForDeletions.Done()
|
||||
}()
|
||||
}
|
||||
|
||||
for {
|
||||
if id, ok := <-idChan; ok {
|
||||
start := time.Now()
|
||||
|
@ -180,16 +202,7 @@ func writeFiles(idChan chan int, fileIdLineChan chan string, s *stats) {
|
|||
if _, err := fp.Upload(0, *b.server); err == nil {
|
||||
if rand.Intn(100) < *b.deletePercentage {
|
||||
s.total++
|
||||
go func() {
|
||||
time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
|
||||
serverLimitChan[fp.Server] <- true
|
||||
if e := operation.DeleteFile(*b.server, fp.Fid); e == nil {
|
||||
s.completed++
|
||||
} else {
|
||||
s.failed++
|
||||
}
|
||||
<-serverLimitChan[fp.Server]
|
||||
}()
|
||||
deleteChan <- fp
|
||||
} else {
|
||||
fileIdLineChan <- fp.Fid
|
||||
}
|
||||
|
@ -211,8 +224,8 @@ func writeFiles(idChan chan int, fileIdLineChan chan string, s *stats) {
|
|||
break
|
||||
}
|
||||
}
|
||||
//wait for the deleting goroutines
|
||||
time.Sleep(time.Duration(1500) * time.Millisecond)
|
||||
close(deleteChan)
|
||||
waitForDeletions.Wait()
|
||||
wait.Done()
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in a new issue