forked from pingcap/tidb
-
Notifications
You must be signed in to change notification settings - Fork 1
/
reorg.go
146 lines (127 loc) · 3.83 KB
/
reorg.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
// Copyright 2015 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.
package ddl
import (
"sync/atomic"
"time"
log "github.com/Sirupsen/logrus"
"github.com/juju/errors"
"github.com/pingcap/tidb/context"
"github.com/pingcap/tidb/kv"
"github.com/pingcap/tidb/meta"
"github.com/pingcap/tidb/model"
"github.com/pingcap/tidb/mysql"
"github.com/pingcap/tidb/util/mock"
)
// newContext gets a context. It is only used for adding column in reorganization state.
func (d *ddl) newContext() context.Context {
c := mock.NewContext()
c.Store = d.store
c.GetSessionVars().SetStatusFlag(mysql.ServerStatusAutocommit, false)
return c
}
const waitReorgTimeout = 10 * time.Second
func (d *ddl) setReorgRowCount(count int64) {
atomic.StoreInt64(&d.reorgRowCount, count)
}
func (d *ddl) getReorgRowCount() int64 {
return atomic.LoadInt64(&d.reorgRowCount)
}
func (d *ddl) runReorgJob(job *model.Job, f func() error) error {
if d.reorgDoneCh == nil {
// start a reorganization job
d.wait.Add(1)
d.reorgDoneCh = make(chan error, 1)
go func() {
defer d.wait.Done()
d.reorgDoneCh <- f()
}()
}
waitTimeout := waitReorgTimeout
// if d.lease is 0, we are using a local storage,
// and we can wait the reorganization to be done here.
// if d.lease > 0, we don't need to wait here because
// we will wait 2 * lease outer and try checking again,
// so we use a very little timeout here.
if d.lease > 0 {
waitTimeout = 1 * time.Millisecond
}
// wait reorganization job done or timeout
select {
case err := <-d.reorgDoneCh:
log.Info("[ddl] run reorg job done")
d.reorgDoneCh = nil
// Update a job's RowCount.
job.SetRowCount(d.getReorgRowCount())
d.setReorgRowCount(0)
return errors.Trace(err)
case <-d.quitCh:
log.Info("[ddl] run reorg job ddl quit")
d.setReorgRowCount(0)
// We return errWaitReorgTimeout here too, so that outer loop will break.
return errWaitReorgTimeout
case <-time.After(waitTimeout):
log.Infof("[ddl] run reorg job wait timeout %v", waitTimeout)
// Update a job's RowCount.
job.SetRowCount(d.getReorgRowCount())
// If timeout, we will return, check the owner and retry to wait job done again.
return errWaitReorgTimeout
}
}
func (d *ddl) isReorgRunnable() error {
if d.isClosed() {
// worker is closed, can't run reorganization.
return errInvalidWorker.Gen("worker is closed")
}
if !d.isOwner() {
// If it's not the owner, we will try later, so here just returns an error.
log.Infof("[ddl] the %s not the job owner", d.uuid)
return errors.Trace(errNotOwner)
}
return nil
}
type reorgInfo struct {
*model.Job
Handle int64
d *ddl
first bool
}
func (d *ddl) getReorgInfo(t *meta.Meta, job *model.Job) (*reorgInfo, error) {
var err error
info := &reorgInfo{
Job: job,
d: d,
first: job.SnapshotVer == 0,
}
if info.first {
// get the current version for reorganization if we don't have
var ver kv.Version
ver, err = d.store.CurrentVersion()
if err != nil {
return nil, errors.Trace(err)
} else if ver.Ver <= 0 {
return nil, errInvalidStoreVer.Gen("invalid storage current version %d", ver.Ver)
}
job.SnapshotVer = ver.Ver
} else {
info.Handle, err = t.GetDDLReorgHandle(job)
if err != nil {
return nil, errors.Trace(err)
}
}
return info, errors.Trace(err)
}
func (r *reorgInfo) UpdateHandle(txn kv.Transaction, handle int64) error {
t := meta.NewMeta(txn)
return errors.Trace(t.UpdateDDLReorgHandle(r.Job, handle))
}