etcd mutex блокировка с помощью ttl - PullRequest
Новая
       7

etcd mutex блокировка с помощью ttl

0 голосов
/ 04 октября 2018

Я пытаюсь создать простую клиентскую программу etcd golang etcd, которая использует мьютекс etcd для создания общей блокировки с тайм-аутом.Цель состоит в том, чтобы истек срок действия мьютекса через некоторое время. 

package main

import (
    "context"
    "log"
    "time"

    "go.etcd.io/etcd/clientv3"
    "go.etcd.io/etcd/clientv3/concurrency"
)

var c chan int

func init() {
    c = make(chan int)
}

func main() {
    client, err := clientv3.New(clientv3.Config{
        Endpoints: []string{"http://localhost:2379"},
    })
    if err != nil {
        panic(err)
    }

    watcher := clientv3.NewWatcher(client)
    channel := watcher.Watch(context.Background(), "/foobar", clientv3.WithPrefix())
    go func() {
        for {
            select {
            case change := <-channel:
                for _, ev := range change.Events {
                    log.Printf("etcd change on key; %s, type = %v", string(ev.Kv.Key), ev.Type)
                }
            }
        }
    }()

    go lockFoobar(client, 1)
    go lockFoobar(client, 2)
    <-c
    <-c
}

func lockFoobar(client *clientv3.Client, id int) {
    res, err := client.Grant(context.Background(), 1)
    if err != nil {
        panic(err)
    }

    session, err := concurrency.NewSession(client, concurrency.WithLease(res.ID))
    if err != nil {
        panic(err)
    }

    mux := concurrency.NewMutex(session, "/foobar")

    log.Printf("trying to lock by #%d\n", id)
    ctx, _ := context.WithTimeout(context.Background(), 15*time.Second)
    if err := mux.Lock(ctx); err != nil {
        log.Printf("failed to lock #%d: %v\n", id, err)
        c <- id
        return
    }

    log.Printf("post-lock #%d (lease ID = %x) bullshit\n", id, res.ID)
    time.Sleep(10 * time.Second)
    ttl, _ := client.TimeToLive(context.TODO(), res.ID)
    log.Printf("post-post-lock-#%d-sleep. lease ttl = %v", id, ttl.TTL)
    // mux.Unlock(ctx)
    // log.Printf("post-unlock #%d bullshit\n", id)

    time.Sleep(200 * time.Millisecond)
    c <- id
}

Срок аренды равен 1 секунде, в то время как контекст имеет тайм-аут 5 секунд, поэтому блокировка должна была быть удалена к тому времениконтекст истекает.Однако «заблокированная» блокировка всегда удаляется только после неудачной блокировки, независимо от времени ожидания контекста.

Это текущий вывод: 

2018-10-04 18:39:59.413274 I | trying to lock by #2
2018-10-04 18:39:59.414530 I | trying to lock by #1
2018-10-04 18:39:59.414656 I | etcd change on key; /foobar/2a0966398d0677a2, type = PUT
2018-10-04 18:39:59.414684 I | post-lock #2 (lease ID = 2a0966398d0677a2) bullshit
2018-10-04 18:39:59.415617 I | etcd change on key; /foobar/2a0966398d0677a4, type = PUT
2018-10-04 18:40:10.239045 I | post-post-lock-#2-sleep. lease ttl = 1                       <-- lock for #2 has ttl = 1 even after 10s
2018-10-04 18:40:15.238871 I | failed to lock #1: context deadline exceeded                 <-- lock for #1 fails after 15s

Как видите, блокировка№ 2 все еще жив даже после 15 с.

Запуск ETCDCTL_API=3 etcdctl watch --prefix=true /foobar в другом терминале для наблюдения за изменениями клавиш показывает следующий вывод 

PUT
/foobar/2a0966398d0677a2

PUT
/foobar/2a0966398d0677a4

DELETE
/foobar/2a0966398d0677a4

DELETE
/foobar/2a0966398d0677a2

Это предполагаемое поведение?Есть ли способ выполнить то, что я пытаюсь сделать?

PS: Реальный вариант использования заключается в создании программы, которая запускается в нескольких случаях и не оставляет блокировок в etcd при сбое и / илиубить (SIGKILL).

1 Ответ

0 голосов
/ 05 октября 2018

После некоторого поиска я нашел причину такого поведения.Сеанс поддерживает аренду до тех пор, пока не произойдет ошибка или отмена.

С session.go: 

...
// keep the lease alive until client error or cancelled context
go func() {
    defer close(donec)
    for range keepAlive {
        // eat messages until keep alive channel closes
    }
}()
...

Callint session.Orphan() после создания мьютекса не позволит поддерживать сеанс в рабочем состоянии ислужит моей цели.

...