Function New

xidmap/xidmap.go:90–183

New creates an XidMap. zero conn must be valid for UID allocations to happen. Optionally, a badger.DB can be provided to persist the xid to uid allocations. This would add latency to the assignment operations. XidMap creates the temporary buffers inside dir directory. The caller must ensure that the

(opts XidMapOptions)

// assignment operations. XidMap creates the temporary buffers inside dir directory. The caller must
// ensure that the dir exists.
func New(opts XidMapOptions) *XidMap {
	numShards := 32
	xm := &XidMap{
		newRanges: make(chan *pb.AssignedIds, numShards),
		shards:    make([]*shard, numShards),
		kvChan:    make(chan []kv, 64),
		dg:        opts.DgClient,
	}
	for i := range xm.shards {
		xm.shards[i] = &shard{
			tree: z.NewTree("XidMap"),
		}
	}

	if opts.DB != nil {
		// If DB is provided, let's load up all the xid -> uid mappings in memory.
		xm.writer = opts.DB.NewWriteBatch()

		for range 16 {
			xm.wg.Add(1)
			go xm.dbWriter()
		}

		err := opts.DB.View(func(txn *badger.Txn) error {
			var count int
			opt := badger.DefaultIteratorOptions
			opt.PrefetchValues = false
			itr := txn.NewIterator(opt)
			defer itr.Close()
			for itr.Rewind(); itr.Valid(); itr.Next() {
				item := itr.Item()
				key := string(item.Key())
				sh := xm.shardFor(key)
				err := item.Value(func(val []byte) error {
					uid := binary.BigEndian.Uint64(val)
					// No need to acquire a lock. This is all serial access.
					sh.tree.Set(farm.Fingerprint64([]byte(key)), uid)
					return nil
				})
				if err != nil {
					return err
				}
				count++
			}
			glog.Infof("Loaded up %d xid to uid mappings", count)
			return nil
		})
		x.Check(err)
	}

	if opts.UidAssigner != nil {
		xm.zc = pb.NewZeroClient(opts.UidAssigner)
	}

	go func() {
		const initBackoff = 10 * time.Millisecond
		const maxBackoff = 5 * time.Second
		backoff := initBackoff