Merge pull request #10 from ibuildthecloud/master

Fix pessimestic update revision check

Author: ibuildthecloud

pkg/drivers/generic/generic.go

@@ -294,8 +294,11 @@ func (d *Generic) CurrentRevision(ctx context.Context) (int64, error) {
 	return id, err
 }
 
-func (d *Generic) After(ctx context.Context, prefix string, rev int64) (*sql.Rows, error) {
+func (d *Generic) After(ctx context.Context, prefix string, rev, limit int64) (*sql.Rows, error) {
 	sql := d.AfterSQL
+	if limit > 0 {
+		sql = fmt.Sprintf("%s LIMIT %d", sql, limit)
+	}
 	return d.query(ctx, sql, prefix, rev)
 }
 

pkg/logstructured/logstructured.go

@@ -13,7 +13,7 @@ type Log interface {
 	Start(ctx context.Context) error
 	CurrentRevision(ctx context.Context) (int64, error)
 	List(ctx context.Context, prefix, startKey string, limit, revision int64, includeDeletes bool) (int64, []*server.Event, error)
-	After(ctx context.Context, prefix string, revision int64) (int64, []*server.Event, error)
+	After(ctx context.Context, prefix string, revision, limit int64) (int64, []*server.Event, error)
 	Watch(ctx context.Context, prefix string) <-chan []*server.Event
 	Count(ctx context.Context, prefix string) (int64, int64, error)
 	Append(ctx context.Context, event *server.Event) (int64, error)
@@ -185,10 +185,14 @@ func (l *LogStructured) Count(ctx context.Context, prefix string) (revRet int64,
 func (l *LogStructured) Update(ctx context.Context, key string, value []byte, revision, lease int64) (revRet int64, kvRet *server.KeyValue, updateRet bool, errRet error) {
 	defer func() {
 		l.adjustRevision(ctx, &revRet)
-		logrus.Debugf("UPDATE %s, value=%d, rev=%d, lease=%v => rev=%d, kv=%v, updated=%v, err=%v", key, len(value), revision, lease, revRet, kvRet != nil, updateRet, errRet)
+		kvRev := int64(0)
+		if kvRet != nil {
+			kvRev = kvRet.ModRevision
+		}
+		logrus.Debugf("UPDATE %s, value=%d, rev=%d, lease=%v => rev=%d, kvrev=%d, updated=%v, err=%v", key, len(value), revision, lease, revRet, kvRev, updateRet, errRet)
 	}()
 
-	rev, event, err := l.get(ctx, key, revision, false)
+	rev, event, err := l.get(ctx, key, 0, false)
 	if err != nil {
 		return 0, nil, false, err
 	}
@@ -295,7 +299,7 @@ func (l *LogStructured) Watch(ctx context.Context, prefix string, revision int64
 
 	result := make(chan []*server.Event, 100)
 
-	rev, kvs, err := l.log.After(ctx, prefix, revision)
+	rev, kvs, err := l.log.After(ctx, prefix, revision, 0)
 	if err != nil {
 		logrus.Errorf("failed to list %s for revision %d", prefix, revision)
 		cancel()

pkg/logstructured/sqllog/sql.go

@@ -31,7 +31,7 @@ type Dialect interface {
 	List(ctx context.Context, prefix, startKey string, limit, revision int64, includeDeleted bool) (*sql.Rows, error)
 	Count(ctx context.Context, prefix string) (int64, int64, error)
 	CurrentRevision(ctx context.Context) (int64, error)
-	After(ctx context.Context, prefix string, rev int64) (*sql.Rows, error)
+	After(ctx context.Context, prefix string, rev, limit int64) (*sql.Rows, error)
 	Insert(ctx context.Context, key string, create, delete bool, createRevision, previousRevision int64, ttl int64, value, prevValue []byte) (int64, error)
 	GetRevision(ctx context.Context, revision int64) (*sql.Rows, error)
 	DeleteRevision(ctx context.Context, revision int64) error
@@ -152,12 +152,12 @@ func (s *SQLLog) CurrentRevision(ctx context.Context) (int64, error) {
 	return s.d.CurrentRevision(ctx)
 }
 
-func (s *SQLLog) After(ctx context.Context, prefix string, revision int64) (int64, []*server.Event, error) {
+func (s *SQLLog) After(ctx context.Context, prefix string, revision, limit int64) (int64, []*server.Event, error) {
 	if strings.HasSuffix(prefix, "/") {
 		prefix += "%"
 	}
 
-	rows, err := s.d.After(ctx, prefix, revision)
+	rows, err := s.d.After(ctx, prefix, revision, limit)
 	if err != nil {
 		return 0, nil, err
 	}
@@ -280,27 +280,31 @@ func (s *SQLLog) startWatch() (chan interface{}, error) {
 
 func (s *SQLLog) poll(result chan interface{}, pollStart int64) {
 	var (
-		last     = pollStart
-		skip     int64
-		skipTime time.Time
+		last        = pollStart
+		skip        int64
+		skipTime    time.Time
+		waitForMore = true
 	)
 
 	wait := time.NewTicker(time.Second)
 	defer wait.Stop()
 	defer close(result)
 
 	for {
-		select {
-		case <-s.ctx.Done():
-			return
-		case check := <-s.notify:
-			if check <= last {
-				continue
+		if waitForMore {
+			select {
+			case <-s.ctx.Done():
+				return
+			case check := <-s.notify:
+				if check <= last {
+					continue
+				}
+			case <-wait.C:
 			}
-		case <-wait.C:
 		}
+		waitForMore = true
 
-		rows, err := s.d.After(s.ctx, "%", last)
+		rows, err := s.d.After(s.ctx, "%", last, 500)
 		if err != nil {
 			logrus.Errorf("fail to list latest changes: %v", err)
 			continue
@@ -316,6 +320,8 @@ func (s *SQLLog) poll(result chan interface{}, pollStart int64) {
 			continue
 		}
 
+		waitForMore = len(events) < 100
+
 		rev := last
 		var (
 			sequential []*server.Event

pkg/server/server.go

@@ -104,10 +104,8 @@ func toKVs(kvs ...*KeyValue) []*mvccpb.KeyValue {
 	ret := make([]*mvccpb.KeyValue, 0, len(kvs))
 	for _, kv := range kvs {
 		newKV := toKV(kv)
-		if newKV == nil {
-			fmt.Println("HIHIHIH")
-		} else {
-			ret = append(ret, toKV(kv))
+		if newKV != nil {
+			ret = append(ret, newKV)
 		}
 	}
 	return ret