Repository navigation
v0.2.3-RELEASE
Big Changes: Concurrency and Worker Pool's max Goroutine Numbers
ChangeStateHandler is now concurrently handled.
Affected endpoints: start, stop, pause, resume, delete
Pros: Task handling is MUCH faster
Cons: Difficult code refactoring, need to consider about SW Architecture and HW specs.
See this code:
func ChangeStateHandler(state string) http.HandlerFunc {
return func(wr http.ResponseWriter, req *http.Request) {
tagBytes, err := io.ReadAll(req.Body)
if err != nil {
log.Printf("%s: Failed to read request body: %v", state, err)
http.Error(wr, err.Error(), http.StatusBadRequest)
return
}
Tag := strings.Trim(string(tagBytes), "\"")
log.Printf("%s: Received request to stop container with tag '%s'.", state, Tag)
info := StateChangeTarget{
Tag: Tag,
Status: state,
}
select {
case WorkQueue.StateTasks <- info:
log.Printf("ChangeContainer: Added container %s task to the work queue.\n", state)
wr.WriteHeader(http.StatusOK)
wr.Write([]byte(fmt.Sprintf("%s command sent for container '%s'", state, Tag)))
return
default:
log.Println("ChangeContainer: Work queue is full.")
http.Error(wr, "Server is busy", http.StatusServiceUnavailable)
return
}
}
}This sends Tag:Status pair into WorkQueue.
Previous work queue didn't support StateChangeTarget nor it was actually declared as type.
type ContainerQueue struct {
Tasks chan linux_virt_unit.ContainerInfo
wg sync.WaitGroup
StateTasks chan StateChangeTarget
}Here, I added StateTasks channel to receive tag and target status.
(...)
// Stop stops the worker goroutines.
func (q *ContainerQueue) Stop() {
log.Println("Stop: Stopping worker goroutines.")
close(q.Tasks)
close(q.StateTasks)
q.wg.Wait()
log.Println("Stop: All worker goroutines stopped.")
}Work queue stop function must wait close both channels.
(...)
func (q *ContainerQueue) StateChangeWorker() {
defer q.wg.Done()
log.Println("worker: Worker goroutine started.")
for target := range q.StateTasks {
if target.Status == "delete" {
DeleteContainerByName(target.Tag)
}
err := ChangeState(target.Tag, target.Status)
if err != nil {
log.Printf("Status change failed on task %s, err: %v\n", target.Status, err)
}
}
log.Println("worker: Worker goroutine finished.")
}Added worker needs SEPARATED goroutine. If a developer mash down two goroutine calls into a unified goroutine, one of a pair should be in a broken state;be aware of it.
Also, partially automatic sync in Kivy app, has been released here.
def update_container_list(self, containers):
self.container_list.clear_widgets()
self.selected_containers = {}
placeholder = MDLabel(
text="Container List",
halign='center',
theme_text_color="Hint",
size_hint_y=None,
height=dp(40),
)
self.container_list.add_widget(placeholder)
for container in containers:
tmp = container['tag']
tag_split = tmp.split("-")
tag_split = tag_split[:-1]
containerLabel = ""
for itm in tag_split:
containerLabel += itm
containerLabel += '-'
containerLabel = containerLabel[:-1]
print(containerLabel)
item = ContainerListItem(tag=containerLabel, port=container['serverport'], distro=container['distro'], version=container['version'], status=container['vmstatus'], actualTag = tmp)
self.container_list.add_widget(item)
self.selected_containers[container['tag']] = item
def manage_container(self, action):
selected_items = [item for item in self.container_list.children if isinstance(item, ContainerListItem) and item.checkbox_active]
if not selected_items:
self.feedback_label.text = "Please select at least one container."
return
main_screen = self.manager.get_screen("main")
#See this loop
for item in selected_items:
main_screen.current_selected_tag = item.actualTag
self.manager.get_screen("main").send_request(action)
main_screen.current_selected_tag = None # Reset selection after action
self.list_containers_and_display_json(None)
self.manager.get_screen("main").send_request("request")
main_screen = self.manager.get_screen("main")
prev_list = current_list = main_screen.containers
manage_screen = self.manager.get_screen("manage")
start = time.time()
while prev_list == current_list:
self.manager.get_screen("main").send_request("request")
current_list = self.manager.get_screen("main").containers
end = time.time()
if end-start > 1:
break
manage_screen.update_container_list(current_list)it waits for status change for a second.
Also, update_container_list is called when each request buttons in manage screen has pressed by user.