func (pl *ProcessList) AddTableProgress(pid uint64, name string, total int64) {
	pl.mu.Lock()
	defer pl.mu.Unlock()
	p, ok := pl.procs[pid]
	if !ok {
		return
	}
	if pg, ok := p.Progress[name]; ok {
		pg.Total = total
		p.Progress[name] = pg
	} else {
		p.Progress[name] = NewTableProgress(name, total)
	}
}
AddTableProgress adds a new item to track progress from to the process with
the given pid. If the pid does not exist, it will do nothing.
package sql
import (
	"context"
	"fmt"
	"sync"
	"time"
	"github.com/sirupsen/logrus"
	"gopkg.in/src-d/go-errors.v1"
)
type Progress struct {
	Name  string
	Done  int64
	Total int64
}
func (p Progress) totalString() string {
	var total = "?"
	if p.Total > 0 {
		total = fmt.Sprint(p.Total)
	}
	return total
}
type TableProgress struct {
	Progress
	PartitionsProgress map[string]PartitionProgress
}
func NewTableProgress(name string, total int64) TableProgress {
	return TableProgress{
		Progress: Progress{
			Name:  name,
			Total: total,
		},
		PartitionsProgress: make(map[string]PartitionProgress),
	}
}
func (p TableProgress) String() string {
	return fmt.Sprintf("%s (%d/%s partitions)", p.Name, p.Done, p.totalString())
}
type PartitionProgress struct {
	Progress
}
func (p PartitionProgress) String() string {
	return fmt.Sprintf("%s (%d/%s rows)", p.Name, p.Done, p.totalString())
}
type ProcessType byte
const (
	QueryProcess ProcessType = iota
	CreateIndexProcess
)
func (p ProcessType) String() string {
	switch p {
	case QueryProcess:
		return "query"
	case CreateIndexProcess:
		return "create_index"
	default:
		return "invalid"
	}
}
type Process struct {
	Pid        uint64
	Connection uint32
	User       string
	Type       ProcessType
	Query      string
	Progress   map[string]TableProgress
	StartedAt  time.Time
	Kill       context.CancelFunc
}
func (p *Process) Done() { p.Kill() }
func (p *Process) Seconds() uint64 {
	return uint64(time.Since(p.StartedAt) / time.Second)
}
type ProcessList struct {
	mu    sync.RWMutex
	procs map[uint64]*Process
}
func NewProcessList() *ProcessList {
	return &ProcessList{
		procs: make(map[uint64]*Process),
	}
}
var ErrPidAlreadyUsed = errors.NewKind("pid %d is already in use")
func (pl *ProcessList) AddProcess(
	ctx *Context,
	typ ProcessType,
	query string,
) (*Context, error) {
	pl.mu.Lock()
	defer pl.mu.Unlock()
	if _, ok := pl.procs[ctx.Pid()]; ok {
		return nil, ErrPidAlreadyUsed.New(ctx.Pid())
	}
	newCtx, cancel := context.WithCancel(ctx)
	ctx = ctx.WithContext(newCtx)
	pl.procs[ctx.Pid()] = &Process{
		Pid:        ctx.Pid(),
		Connection: ctx.ID(),
		Type:       typ,
		Query:      query,
		Progress:   make(map[string]TableProgress),
		User:       ctx.Session.Client().User,
		StartedAt:  time.Now(),
		Kill:       cancel,
	}
	return ctx, nil
}
func (pl *ProcessList) UpdateTableProgress(pid uint64, name string, delta int64) {
	pl.mu.Lock()
	defer pl.mu.Unlock()
	p, ok := pl.procs[pid]
	if !ok {
		return
	}
	progress, ok := p.Progress[name]
	if !ok {
		progress = NewTableProgress(name, -1)
	}
	progress.Done += delta
	p.Progress[name] = progress
}
func (pl *ProcessList) UpdatePartitionProgress(pid uint64, tableName, partitionName string, delta int64) {
	pl.mu.Lock()
	defer pl.mu.Unlock()
	p, ok := pl.procs[pid]
	if !ok {
		return
	}
	tablePg, ok := p.Progress[tableName]
	if !ok {
		return
	}
	partitionPg, ok := tablePg.PartitionsProgress[partitionName]
	if !ok {
		partitionPg = PartitionProgress{Progress: Progress{Name: partitionName, Total: -1}}
	}
	partitionPg.Done += delta
	tablePg.PartitionsProgress[partitionName] = partitionPg
}