diff options
Diffstat (limited to 'models')
| -rw-r--r-- | models/pull.go | 5 | ||||
| -rw-r--r-- | models/repo.go | 203 | ||||
| -rw-r--r-- | models/repo_mirror.go | 243 | ||||
| -rw-r--r-- | models/webhook.go | 65 | ||||
| -rw-r--r-- | models/wiki.go | 2 |
5 files changed, 258 insertions, 260 deletions
diff --git a/models/pull.go b/models/pull.go index 64b34755..100d4db4 100644 --- a/models/pull.go +++ b/models/pull.go @@ -20,8 +20,11 @@ import ( "github.com/gogits/gogs/modules/log" "github.com/gogits/gogs/modules/process" "github.com/gogits/gogs/modules/setting" + "github.com/gogits/gogs/modules/sync" ) +var PullRequestQueue = sync.NewUniqueQueue(setting.Repository.PullRequestQueueLength) + type PullRequestType int const ( @@ -537,8 +540,6 @@ func (pr *PullRequest) UpdateCols(cols ...string) error { return err } -var PullRequestQueue = NewUniqueQueue(setting.Repository.PullRequestQueueLength) - // UpdatePatch generates and saves a new patch. func (pr *PullRequest) UpdatePatch() (err error) { if err = pr.GetHeadRepo(); err != nil { diff --git a/models/repo.go b/models/repo.go index a26ba924..932a0e9f 100644 --- a/models/repo.go +++ b/models/repo.go @@ -40,7 +40,7 @@ const ( _TPL_UPDATE_HOOK = "#!/usr/bin/env %s\n%s update $1 $2 $3 --config='%s'\n" ) -var repoWorkingPool = sync.NewSingleInstancePool() +var repoWorkingPool = sync.NewExclusivePool() var ( ErrRepoFileNotExist = errors.New("Repository file does not exist") @@ -381,7 +381,7 @@ func (repo *Repository) IssueStats(uid int64, filterMode int, isPull bool) (int6 } func (repo *Repository) GetMirror() (err error) { - repo.Mirror, err = GetMirror(repo.ID) + repo.Mirror, err = GetMirrorByRepoID(repo.ID) return err } @@ -574,136 +574,6 @@ func (repo *Repository) CloneLink() (cl *CloneLink) { return repo.cloneLink(false) } -// Mirror represents a mirror information of repository. -type Mirror struct { - ID int64 `xorm:"pk autoincr"` - RepoID int64 - Repo *Repository `xorm:"-"` - Interval int // Hour. - EnablePrune bool `xorm:"NOT NULL DEFAULT true"` - - Updated time.Time `xorm:"-"` - UpdatedUnix int64 - NextUpdate time.Time `xorm:"-"` - NextUpdateUnix int64 - - address string `xorm:"-"` -} - -func (m *Mirror) BeforeInsert() { - m.NextUpdateUnix = m.NextUpdate.Unix() -} - -func (m *Mirror) BeforeUpdate() { - m.UpdatedUnix = time.Now().Unix() - m.NextUpdateUnix = m.NextUpdate.Unix() -} - -func (m *Mirror) AfterSet(colName string, _ xorm.Cell) { - var err error - switch colName { - case "repo_id": - m.Repo, err = GetRepositoryByID(m.RepoID) - if err != nil { - log.Error(3, "GetRepositoryByID[%d]: %v", m.ID, err) - } - case "updated_unix": - m.Updated = time.Unix(m.UpdatedUnix, 0).Local() - case "next_updated_unix": - m.NextUpdate = time.Unix(m.NextUpdateUnix, 0).Local() - } -} - -func (m *Mirror) readAddress() { - if len(m.address) > 0 { - return - } - - cfg, err := ini.Load(m.Repo.GitConfigPath()) - if err != nil { - log.Error(4, "Load: %v", err) - return - } - m.address = cfg.Section("remote \"origin\"").Key("url").Value() -} - -// HandleCloneUserCredentials replaces user credentials from HTTP/HTTPS URL -// with placeholder <credentials>. -// It will fail for any other forms of clone addresses. -func HandleCloneUserCredentials(url string, mosaics bool) string { - i := strings.Index(url, "@") - if i == -1 { - return url - } - start := strings.Index(url, "://") - if start == -1 { - return url - } - if mosaics { - return url[:start+3] + "<credentials>" + url[i:] - } - return url[:start+3] + url[i+1:] -} - -// Address returns mirror address from Git repository config without credentials. -func (m *Mirror) Address() string { - m.readAddress() - return HandleCloneUserCredentials(m.address, false) -} - -// FullAddress returns mirror address from Git repository config. -func (m *Mirror) FullAddress() string { - m.readAddress() - return m.address -} - -// SaveAddress writes new address to Git repository config. -func (m *Mirror) SaveAddress(addr string) error { - configPath := m.Repo.GitConfigPath() - cfg, err := ini.Load(configPath) - if err != nil { - return fmt.Errorf("Load: %v", err) - } - - cfg.Section("remote \"origin\"").Key("url").SetValue(addr) - return cfg.SaveToIndent(configPath, "\t") -} - -func getMirror(e Engine, repoId int64) (*Mirror, error) { - m := &Mirror{RepoID: repoId} - has, err := e.Get(m) - if err != nil { - return nil, err - } else if !has { - return nil, ErrMirrorNotExist - } - return m, nil -} - -// GetMirror returns mirror object by given repository ID. -func GetMirror(repoId int64) (*Mirror, error) { - return getMirror(x, repoId) -} - -func updateMirror(e Engine, m *Mirror) error { - _, err := e.Id(m.ID).AllCols().Update(m) - return err -} - -func UpdateMirror(m *Mirror) error { - return updateMirror(x, m) -} - -func DeleteMirrorByRepoID(repoID int64) error { - _, err := x.Delete(&Mirror{RepoID: repoID}) - return err -} - -func createUpdateHook(repoPath string) error { - return git.SetUpdateHook(repoPath, - fmt.Sprintf(_TPL_UPDATE_HOOK, setting.ScriptType, "\""+setting.AppPath+"\"", setting.CustomConf)) -} - type MigrateRepoOptions struct { Name string Description string @@ -839,6 +709,11 @@ func cleanUpMigrateGitConfig(configPath string) error { return nil } +func createUpdateHook(repoPath string) error { + return git.SetUpdateHook(repoPath, + fmt.Sprintf(_TPL_UPDATE_HOOK, setting.ScriptType, "\""+setting.AppPath+"\"", setting.CustomConf)) +} + // Finish migrating repository and/or wiki with things that don't need to be done for mirrors. func CleanUpMigrateInfo(repo *Repository) (*Repository, error) { repoPath := repo.RepoPath() @@ -1748,70 +1623,6 @@ const ( _CHECK_REPOs = "check_repos" ) -// MirrorUpdate checks and updates mirror repositories. -func MirrorUpdate() { - if taskStatusTable.IsRunning(_MIRROR_UPDATE) { - return - } - taskStatusTable.Start(_MIRROR_UPDATE) - defer taskStatusTable.Stop(_MIRROR_UPDATE) - - log.Trace("Doing: MirrorUpdate") - - mirrors := make([]*Mirror, 0, 10) - if err := x.Where("next_update_unix<=?", time.Now().Unix()).Iterate(new(Mirror), func(idx int, bean interface{}) error { - m := bean.(*Mirror) - if m.Repo == nil { - log.Error(4, "Disconnected mirror repository found: %d", m.ID) - return nil - } - - repoPath := m.Repo.RepoPath() - wikiPath := m.Repo.WikiPath() - timeout := time.Duration(setting.Git.Timeout.Mirror) * time.Second - - gitArgs := []string{"remote", "update"} - if m.EnablePrune { - gitArgs = append(gitArgs, "--prune") - } - - if _, stderr, err := process.ExecDir( - timeout, repoPath, fmt.Sprintf("MirrorUpdate: %s", repoPath), - "git", gitArgs...); err != nil { - desc := fmt.Sprintf("Fail to update mirror repository(%s): %s", repoPath, stderr) - log.Error(4, desc) - if err = CreateRepositoryNotice(desc); err != nil { - log.Error(4, "CreateRepositoryNotice: %v", err) - } - return nil - } - if m.Repo.HasWiki() { - if _, stderr, err := process.ExecDir( - timeout, wikiPath, fmt.Sprintf("MirrorUpdate: %s", wikiPath), - "git", "remote", "update", "--prune"); err != nil { - desc := fmt.Sprintf("Fail to update mirror wiki repository(%s): %s", wikiPath, stderr) - log.Error(4, desc) - if err = CreateRepositoryNotice(desc); err != nil { - log.Error(4, "CreateRepositoryNotice: %v", err) - } - return nil - } - } - - m.NextUpdate = time.Now().Add(time.Duration(m.Interval) * time.Hour) - mirrors = append(mirrors, m) - return nil - }); err != nil { - log.Error(4, "MirrorUpdate: %v", err) - } - - for i := range mirrors { - if err := UpdateMirror(mirrors[i]); err != nil { - log.Error(4, "UpdateMirror[%d]: %v", mirrors[i].ID, err) - } - } -} - // GitFsck calls 'git fsck' to check repository health. func GitFsck() { if taskStatusTable.IsRunning(_GIT_FSCK) { diff --git a/models/repo_mirror.go b/models/repo_mirror.go new file mode 100644 index 00000000..53014e9d --- /dev/null +++ b/models/repo_mirror.go @@ -0,0 +1,243 @@ +// Copyright 2016 The Gogs Authors. All rights reserved. +// Use of this source code is governed by a MIT-style +// license that can be found in the LICENSE file. + +package models + +import ( + "fmt" + "strings" + "time" + + "github.com/Unknwon/com" + "github.com/go-xorm/xorm" + "gopkg.in/ini.v1" + + "github.com/gogits/gogs/modules/log" + "github.com/gogits/gogs/modules/process" + "github.com/gogits/gogs/modules/setting" + "github.com/gogits/gogs/modules/sync" +) + +var MirrorQueue = sync.NewUniqueQueue(setting.Repository.MirrorQueueLength) + +// Mirror represents mirror information of a repository. +type Mirror struct { + ID int64 `xorm:"pk autoincr"` + RepoID int64 + Repo *Repository `xorm:"-"` + Interval int // Hour. + EnablePrune bool `xorm:"NOT NULL DEFAULT true"` + + Updated time.Time `xorm:"-"` + UpdatedUnix int64 + NextUpdate time.Time `xorm:"-"` + NextUpdateUnix int64 + + address string `xorm:"-"` +} + +func (m *Mirror) BeforeInsert() { + m.NextUpdateUnix = m.NextUpdate.Unix() +} + +func (m *Mirror) BeforeUpdate() { + m.UpdatedUnix = time.Now().Unix() + m.NextUpdateUnix = m.NextUpdate.Unix() +} + +func (m *Mirror) AfterSet(colName string, _ xorm.Cell) { + var err error + switch colName { + case "repo_id": + m.Repo, err = GetRepositoryByID(m.RepoID) + if err != nil { + log.Error(3, "GetRepositoryByID[%d]: %v", m.ID, err) + } + case "updated_unix": + m.Updated = time.Unix(m.UpdatedUnix, 0).Local() + case "next_updated_unix": + m.NextUpdate = time.Unix(m.NextUpdateUnix, 0).Local() + } +} + +// ScheduleNextUpdate calculates and sets next update time. +func (m *Mirror) ScheduleNextUpdate() { + m.NextUpdate = time.Now().Add(time.Duration(m.Interval) * time.Hour) +} + +func (m *Mirror) readAddress() { + if len(m.address) > 0 { + return + } + + cfg, err := ini.Load(m.Repo.GitConfigPath()) + if err != nil { + log.Error(4, "Load: %v", err) + return + } + m.address = cfg.Section("remote \"origin\"").Key("url").Value() +} + +// HandleCloneUserCredentials replaces user credentials from HTTP/HTTPS URL +// with placeholder <credentials>. +// It will fail for any other forms of clone addresses. +func HandleCloneUserCredentials(url string, mosaics bool) string { + i := strings.Index(url, "@") + if i == -1 { + return url + } + start := strings.Index(url, "://") + if start == -1 { + return url + } + if mosaics { + return url[:start+3] + "<credentials>" + url[i:] + } + return url[:start+3] + url[i+1:] +} + +// Address returns mirror address from Git repository config without credentials. +func (m *Mirror) Address() string { + m.readAddress() + return HandleCloneUserCredentials(m.address, false) +} + +// FullAddress returns mirror address from Git repository config. +func (m *Mirror) FullAddress() string { + m.readAddress() + return m.address +} + +// SaveAddress writes new address to Git repository config. +func (m *Mirror) SaveAddress(addr string) error { + configPath := m.Repo.GitConfigPath() + cfg, err := ini.Load(configPath) + if err != nil { + return fmt.Errorf("Load: %v", err) + } + + cfg.Section("remote \"origin\"").Key("url").SetValue(addr) + return cfg.SaveToIndent(configPath, "\t") +} + +// runSync returns true if sync finished without error. +func (m *Mirror) runSync() bool { + repoPath := m.Repo.RepoPath() + wikiPath := m.Repo.WikiPath() + timeout := time.Duration(setting.Git.Timeout.Mirror) * time.Second + + gitArgs := []string{"remote", "update"} + if m.EnablePrune { + gitArgs = append(gitArgs, "--prune") + } + + if _, stderr, err := process.ExecDir( + timeout, repoPath, fmt.Sprintf("runSync: %s", repoPath), + "git", gitArgs...); err != nil { + desc := fmt.Sprintf("Fail to update mirror repository '%s': %s", repoPath, stderr) + log.Error(4, desc) + if err = CreateRepositoryNotice(desc); err != nil { + log.Error(4, "CreateRepositoryNotice: %v", err) + } + return false + } + if m.Repo.HasWiki() { + if _, stderr, err := process.ExecDir( + timeout, wikiPath, fmt.Sprintf("runSync: %s", wikiPath), + "git", "remote", "update", "--prune"); err != nil { + desc := fmt.Sprintf("Fail to update mirror wiki repository '%s': %s", wikiPath, stderr) + log.Error(4, desc) + if err = CreateRepositoryNotice(desc); err != nil { + log.Error(4, "CreateRepositoryNotice: %v", err) + } + return false + } + } + + return true +} + +func getMirrorByRepoID(e Engine, repoID int64) (*Mirror, error) { + m := &Mirror{RepoID: repoID} + has, err := e.Get(m) + if err != nil { + return nil, err + } else if !has { + return nil, ErrMirrorNotExist + } + return m, nil +} + +// GetMirrorByRepoID returns mirror information of a repository. +func GetMirrorByRepoID(repoID int64) (*Mirror, error) { + return getMirrorByRepoID(x, repoID) +} + +func updateMirror(e Engine, m *Mirror) error { + _, err := e.Id(m.ID).AllCols().Update(m) + return err +} + +func UpdateMirror(m *Mirror) error { + return updateMirror(x, m) +} + +func DeleteMirrorByRepoID(repoID int64) error { + _, err := x.Delete(&Mirror{RepoID: repoID}) + return err +} + +// MirrorUpdate checks and updates mirror repositories. +func MirrorUpdate() { + if taskStatusTable.IsRunning(_MIRROR_UPDATE) { + return + } + taskStatusTable.Start(_MIRROR_UPDATE) + defer taskStatusTable.Stop(_MIRROR_UPDATE) + + log.Trace("Doing: MirrorUpdate") + + if err := x.Where("next_update_unix<=?", time.Now().Unix()).Iterate(new(Mirror), func(idx int, bean interface{}) error { + m := bean.(*Mirror) + if m.Repo == nil { + log.Error(4, "Disconnected mirror repository found: %d", m.ID) + return nil + } + + MirrorQueue.Add(m.RepoID) + return nil + }); err != nil { + log.Error(4, "MirrorUpdate: %v", err) + } +} + +// SyncMirrors checks and syncs mirrors. +// TODO: sync more mirrors at same time. +func SyncMirrors() { + // Start listening on new sync requests. + for repoID := range MirrorQueue.Queue() { + log.Trace("SyncMirrors [repo_id: %v]", repoID) + MirrorQueue.Remove(repoID) + + m, err := GetMirrorByRepoID(com.StrTo(repoID).MustInt64()) + if err != nil { + log.Error(4, "GetMirrorByRepoID [%d]: %v", repoID, err) + continue + } + + if !m.runSync() { + continue + } + + m.ScheduleNextUpdate() + if err = UpdateMirror(m); err != nil { + log.Error(4, "UpdateMirror [%d]: %v", repoID, err) + continue + } + } +} + +func InitSyncMirrors() { + go SyncMirrors() +} diff --git a/models/webhook.go b/models/webhook.go index 2db02741..084a7ee7 100644 --- a/models/webhook.go +++ b/models/webhook.go @@ -10,10 +10,8 @@ import ( "fmt" "io/ioutil" "strings" - "sync" "time" - "github.com/Unknwon/com" "github.com/go-xorm/xorm" gouuid "github.com/satori/go.uuid" @@ -22,8 +20,11 @@ import ( "github.com/gogits/gogs/modules/httplib" "github.com/gogits/gogs/modules/log" "github.com/gogits/gogs/modules/setting" + "github.com/gogits/gogs/modules/sync" ) +var HookQueue = sync.NewUniqueQueue(setting.Webhook.QueueLength) + type HookContentType int const ( @@ -500,64 +501,6 @@ func PrepareWebhooks(repo *Repository, event HookEventType, p api.Payloader) err return nil } -// UniqueQueue represents a queue that guarantees only one instance of same ID is in the line. -type UniqueQueue struct { - lock sync.Mutex - ids map[string]bool - - queue chan string -} - -func (q *UniqueQueue) Queue() <-chan string { - return q.queue -} - -func NewUniqueQueue(queueLength int) *UniqueQueue { - if queueLength <= 0 { - queueLength = 100 - } - - return &UniqueQueue{ - ids: make(map[string]bool), - queue: make(chan string, queueLength), - } -} - -func (q *UniqueQueue) Remove(id interface{}) { - q.lock.Lock() - defer q.lock.Unlock() - delete(q.ids, com.ToStr(id)) -} - -func (q *UniqueQueue) AddFunc(id interface{}, fn func()) { - newid := com.ToStr(id) - - if q.Exist(id) { - return - } - - q.lock.Lock() - q.ids[newid] = true - if fn != nil { - fn() - } - q.lock.Unlock() - q.queue <- newid -} - -func (q *UniqueQueue) Add(id interface{}) { - q.AddFunc(id, nil) -} - -func (q *UniqueQueue) Exist(id interface{}) bool { - q.lock.Lock() - defer q.lock.Unlock() - - return q.ids[com.ToStr(id)] -} - -var HookQueue = NewUniqueQueue(setting.Webhook.QueueLength) - func (t *HookTask) deliver() { t.IsDelivered = true @@ -654,7 +597,7 @@ func DeliverHooks() { // Start listening on new hook requests. for repoID := range HookQueue.Queue() { - log.Trace("DeliverHooks [%v]: processing delivery hooks", repoID) + log.Trace("DeliverHooks [repo_id: %v]", repoID) HookQueue.Remove(repoID) tasks = make([]*HookTask, 0, 5) diff --git a/models/wiki.go b/models/wiki.go index f62cd2ac..bc8aaf66 100644 --- a/models/wiki.go +++ b/models/wiki.go @@ -21,7 +21,7 @@ import ( "github.com/gogits/gogs/modules/sync" ) -var wikiWorkingPool = sync.NewSingleInstancePool() +var wikiWorkingPool = sync.NewExclusivePool() // ToWikiPageURL formats a string to corresponding wiki URL name. func ToWikiPageURL(name string) string { |