WIP: fix payment channel locking

gen/main.go

@@ -34,7 +34,7 @@ func main() {
 
 	err = gen.WriteMapEncodersToFile("./paychmgr/cbor_gen.go", "paychmgr",
 		paychmgr.VoucherInfo{},
-		paychmgr.ChannelInfo{},
+		paychmgr.ChannelInfoStorable{},
 	)
 	if err != nil {
 		fmt.Println(err)

node/builder.go

@@ -107,6 +107,7 @@ const (
 	HandleIncomingMessagesKey
 
 	RegisterClientValidatorKey
+	HandlePaymentChannelManagerKey
 
 	// miner
 	GetParamsKey
@@ -272,6 +273,7 @@ func Online() Option {
 
 			Override(new(*paychmgr.Store), paychmgr.NewStore),
 			Override(new(*paychmgr.Manager), paychmgr.NewManager),
+			Override(HandlePaymentChannelManagerKey, paychmgr.HandleManager),
 			Override(new(*market.FundMgr), market.NewFundMgr),
 			Override(SettlePaymentChannelsKey, settler.SettlePaymentChannels),
 		),

paychmgr/accessorcache.go

@@ -0,0 +1,67 @@
+package paychmgr
+
+import "github.com/filecoin-project/go-address"
+
+// accessorByFromTo gets a channel accessor for a given from / to pair.
+// The channel accessor facilitates locking a channel so that operations
+// must be performed sequentially on a channel (but can be performed at
+// the same time on different channels).
+func (pm *Manager) accessorByFromTo(from address.Address, to address.Address) (*channelAccessor, error) {
+	key := pm.accessorCacheKey(from, to)
+
+	// First take a read lock and check the cache
+	pm.lk.RLock()
+	ca, ok := pm.channels[key]
+	pm.lk.RUnlock()
+	if ok {
+		return ca, nil
+	}
+
+	// Not in cache, so take a write lock
+	pm.lk.Lock()
+	defer pm.lk.Unlock()
+
+	// Need to check cache again in case it was updated between releasing read
+	// lock and taking write lock
+	ca, ok = pm.channels[key]
+	if !ok {
+		// Not in cache, so create a new one and store in cache
+		ca = pm.addAccessorToCache(from, to)
+	}
+
+	return ca, nil
+}
+
+// accessorByAddress gets a channel accessor for a given channel address.
+// The channel accessor facilitates locking a channel so that operations
+// must be performed sequentially on a channel (but can be performed at
+// the same time on different channels).
+func (pm *Manager) accessorByAddress(ch address.Address) (*channelAccessor, error) {
+	// Get the channel from / to
+	pm.lk.RLock()
+	channelInfo, err := pm.store.ByAddress(ch)
+	pm.lk.RUnlock()
+	if err != nil {
+		return nil, err
+	}
+
+	// TODO: cache by channel address so we can get by address instead of using from / to
+	return pm.accessorByFromTo(channelInfo.Control, channelInfo.Target)
+}
+
+// accessorCacheKey returns the cache key use to reference a channel accessor
+func (pm *Manager) accessorCacheKey(from address.Address, to address.Address) string {
+	return from.String() + "->" + to.String()
+}
+
+// addAccessorToCache adds a channel accessor to a cache. Note that channelInfo
+// may be nil if the channel hasn't been created yet, but we still want to
+// reference the same channel accessor for a given from/to, so that all
+// attempts to access a channel use the same lock (the lock on the accessor)
+func (pm *Manager) addAccessorToCache(from address.Address, to address.Address) *channelAccessor {
+	key := pm.accessorCacheKey(from, to)
+	ca := newChannelAccessor(pm)
+	// TODO: Use LRU
+	pm.channels[key] = ca
+	return ca
+}