opt: divide each thread and make them fetch task info individualy

Author: HFO4

api/api.go

@@ -6,13 +6,15 @@ import (
 	"log"
 	"net/http"
 	"strconv"
+	"sync"
 )
 
 //ApiInfo Api请求配置
 type ApiInfo struct {
 	TOKEN    string
 	APIURL   string
 	BASEPATH string
+	Lock     *sync.Mutex
 }
 
 //APIResponse API响应结果
@@ -31,6 +33,11 @@ func (apiInfo *ApiInfo) GetTaskList(num int) string {
 	return apiInfo.apiGet("getList?num=" + strconv.Itoa(num))
 }
 
+//CancelUploadSession 取消分片上传
+func (apiInfo *ApiInfo) CancelUploadSession(url string) {
+	apiInfo.apiDelete(url)
+}
+
 //GetPolicy 获取上传策略详情
 func (apiInfo *ApiInfo) GetPolicy(id int) string {
 	return apiInfo.apiGet("getPolicy?id=" + strconv.Itoa(id))
@@ -50,10 +57,38 @@ func (apiInfo *ApiInfo) SetSuccess(id int) string {
 	return ""
 }
 
+//SetError 设置为失败状态
 func (apiInfo *ApiInfo) SetError(id int) string {
 	return apiInfo.apiGet("setError?id=" + strconv.Itoa(id))
 }
 
+//apiDelete 发送DELETE请求
+
+func (apiInfo *ApiInfo) apiDelete(controller string) string {
+	client := &http.Client{}
+	request, err := http.NewRequest("DELETE", controller, nil)
+	if err != nil {
+		log.Printf("[ERROR] Failed to create DELETE requetst, #%v ", err)
+	}
+	request.Header.Set("Authorization", "Bearer "+apiInfo.TOKEN)
+	response, err := client.Do(request)
+	if err != nil {
+		log.Printf("[ERROR] Failed to send GET requetst, #%v ", err)
+		return ""
+	}
+	defer response.Body.Close()
+	if response.StatusCode == 200 || response.StatusCode == 204 {
+		r, err := ioutil.ReadAll(response.Body)
+		if err != nil {
+			log.Printf("[ERROR] Failed to get GET requetst body, #%v ", err)
+		}
+		return string(r)
+	} else {
+		log.Printf("[ERROR] Failed to get respond, HTTP ERROR %v ", response.StatusCode)
+		return ""
+	}
+}
+
 //apiGet 发送GET请求
 func (apiInfo *ApiInfo) apiGet(controller string) string {
 	client := &http.Client{}

conf.yaml

@@ -1,4 +1,4 @@
 token: 52121225
 api: http://127.0.0.1/Queue
 taskNum: 4
-threadNum: 4
+Duration: 1

main.go

@@ -6,6 +6,8 @@ import (
 	"fmt"
 	"io/ioutil"
 	"log"
+	"sync"
+	"time"
 
 	"./api"
 	"./task"
@@ -14,9 +16,10 @@ import (
 )
 
 type taskConfig struct {
-	TOKEN   string `yaml:"token"`
-	APIURL  string `yaml:"api"`
-	TASKNUM int    `yaml:"taskNum"`
+	TOKEN    string `yaml:"token"`
+	APIURL   string `yaml:"api"`
+	TASKNUM  int    `yaml:"taskNum"`
+	DURATION int    `yaml:"Duration"`
 }
 
 func (c *taskConfig) getConf() (*taskConfig, error) {
@@ -42,23 +45,48 @@ func main() {
 	var config taskConfig
 	_, err := config.getConf()
 	if err == nil {
+
 		log.Printf("[INFO] Config information:  %v ", config)
-		api := api.ApiInfo{TOKEN: config.TOKEN, APIURL: config.APIURL}
+		api := api.ApiInfo{
+			TOKEN:  config.TOKEN,
+			APIURL: config.APIURL,
+			Lock:   new(sync.Mutex),
+		}
 		basicInfo := api.GetBasicInfo()
+
 		if basicInfo != "" {
+
 			log.Printf("[INFO] Basic Info:  %v ", basicInfo)
 			var siteInfo map[string]string
 			err := json.Unmarshal([]byte(basicInfo), &siteInfo)
 			if err != nil {
 				log.Printf("[ERROR] Failed to decode basic infomation,  %v ", err.Error())
 			}
-			for {
-				taskListContent := api.GetTaskList(config.TASKNUM)
-				if taskListContent != "none" {
-					task.Init(taskListContent, api, siteInfo)
-				}
-				break
+
+			var wg sync.WaitGroup
+			for i := 0; i < config.TASKNUM; i++ {
+				wg.Add(1)
+				log.Printf("[Info] Thread %d start", i+1)
+				threadID := i
+				go func() {
+					for {
+
+						api.Lock.Lock()
+						taskListContent := api.GetTaskList(1)
+						api.Lock.Unlock()
+
+						if taskListContent != "none" {
+							task.Init(taskListContent, api, siteInfo, threadID)
+						}
+						time.Sleep(time.Duration(config.DURATION) * time.Second)
+					}
+
+				}()
+				time.Sleep(time.Duration(1) * time.Second)
 			}
+
+			wg.Wait()
+
 		}
 
 	}

task/TaskAdapter.go

@@ -107,8 +107,7 @@ func (task *OneDriveUpload) uploadChunks(Client *onedrive.Client) {
 			uploaded += (v.To - v.From + 1)
 			task.Log(fmt.Sprintf("[Info] Chunk uploaded, From:%d To:%d Total:%d Complete:%.2f", v.From, v.To, task.Attr.Fsize, float32(uploaded)/float32(task.Attr.Fsize)))
 		} else {
-			//to-do:关闭上传Session
-
+			task.Info.apiInfo.CancelUploadSession(url)
 			return
 		}
 	}
@@ -137,13 +136,12 @@ func (task *OneDriveUpload) uploadSingleChunk(chunk Chunk, Client *onedrive.Clie
 		return false
 	}
 
-	res, uploadErr := Client.UploadChunk(url, chunk.From, chunk.To, int(task.Attr.Fsize), r)
+	_, uploadErr := Client.UploadChunk(url, chunk.From, chunk.To, int(task.Attr.Fsize), r)
 	if uploadErr != "" {
 		task.Log("[Error] Failed to upload chunk," + uploadErr)
 		task.Error()
 		return false
 	}
-	fmt.Println(res)
 	r.Close()
 	return true
 }

task/TaskPool.go

@@ -3,7 +3,6 @@ package task
 import (
 	"encoding/json"
 	"log"
-	"sync"
 
 	"../api"
 )
@@ -19,19 +18,16 @@ type SingleTaskInfo struct {
 }
 
 //Init 初始化任务线程池
-func Init(taskListContent string, apiInfo api.ApiInfo, siteInfo map[string]string) {
+func Init(taskListContent string, apiInfo api.ApiInfo, siteInfo map[string]string, threadID int) {
 	var taskStringList []SingleTaskInfo
 	err := json.Unmarshal([]byte(taskListContent), &taskStringList)
 	if err != nil {
 		log.Printf("[ERROR] Failed to decode basic infomation,  %v ", err.Error())
 	}
-	var wg sync.WaitGroup
 	for _, v := range taskStringList {
-		wg.Add(1)
+		log.Printf("[Info][Thread %d] New task: %s", threadID, v.TaskName)
 		signleTask := taskInfo{sqlInfo: v, apiInfo: apiInfo, siteInfo: siteInfo}
-		go func(t taskInfo) {
-			t.Run(&wg)
-		}(signleTask)
+		signleTask.Run()
+
 	}
-	wg.Wait()
 }

task/task.go

@@ -1,8 +1,6 @@
 package task
 
 import (
-	"sync"
-
 	"../api"
 )
 
@@ -41,7 +39,7 @@ type Task interface {
 }
 
 //Run 根据任务类型处理任务
-func (task *taskInfo) Run(wg *sync.WaitGroup) {
+func (task *taskInfo) Run() {
 	var newTask Task
 	switch {
 	case task.sqlInfo.TaskType == "uploadSingleToOnedrive":
@@ -62,5 +60,4 @@ func (task *taskInfo) Run(wg *sync.WaitGroup) {
 		newTask.Init()
 	}
 	newTask.Excute()
-	wg.Done()
 }