Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 1 | /* |
| 2 | * Copyright 2018-present Open Networking Foundation |
| 3 | * |
| 4 | * Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | * you may not use this file except in compliance with the License. |
| 6 | * You may obtain a copy of the License at |
| 7 | * |
| 8 | * http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | * |
| 10 | * Unless required by applicable law or agreed to in writing, software |
| 11 | * distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | * See the License for the specific language governing permissions and |
| 14 | * limitations under the License. |
| 15 | */ |
| 16 | |
| 17 | /* |
| 18 | * Two voltha cores receive the same request; each tries to acquire ownership of the request |
| 19 | * by writing its identifier (e.g. container name or pod name) to the transaction key named |
Richard Jankowski | e4d7766 | 2018-10-17 13:53:21 -0400 | [diff] [blame] | 20 | * after the serial number of the request. The core that loses the race for acquisition |
| 21 | * monitors the progress of the core actually serving the request by watching for changes |
| 22 | * in the value of the transaction key. Once the request is complete, the |
| 23 | * serving core closes the transaction by invoking the KVTransaction's Close method, which |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 24 | * replaces the value of the transaction (i.e. serial number) key with the string |
| 25 | * TRANSACTION_COMPLETE. The standby core observes this update, stops watching the transaction, |
| 26 | * and then deletes the transaction key. |
| 27 | * |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 28 | */ |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 29 | |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 30 | package core |
| 31 | |
| 32 | import ( |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 33 | "context" |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 34 | "time" |
| 35 | |
serkant.uluderya | 2ae470f | 2020-01-21 11:13:09 -0800 | [diff] [blame] | 36 | "github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore" |
| 37 | "github.com/opencord/voltha-lib-go/v3/pkg/log" |
Kent Hagerman | 46dcd9d | 2019-09-18 16:42:59 -0400 | [diff] [blame] | 38 | "google.golang.org/grpc/codes" |
| 39 | "google.golang.org/grpc/status" |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 40 | ) |
| 41 | |
| 42 | // Transaction acquisition results |
| 43 | const ( |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 44 | UNKNOWN = iota |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 45 | SeizedBySelf |
| 46 | CompletedByOther |
| 47 | AbandonedByOther |
| 48 | AbandonedWatchBySelf |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 49 | ) |
| 50 | |
Kent Hagerman | 46dcd9d | 2019-09-18 16:42:59 -0400 | [diff] [blame] | 51 | var errorTransactionNotAcquired = status.Error(codes.Canceled, "transaction-not-acquired") |
| 52 | |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 53 | // Transaction constant |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 54 | const ( |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 55 | TransactionComplete = "TRANSACTION-COMPLETE" |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 56 | ) |
| 57 | |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 58 | // Transaction constants used to guarantee the Core processing a request hold on to the transaction until |
khenaidoo | f684e1b | 2019-10-28 19:00:37 -0400 | [diff] [blame] | 59 | // it either completes it (either successfully or times out) or the Core itself crashes ( |
| 60 | // e.g. a server failure). |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 61 | // If a request has a timeout of x seconds then the Core processing the request will renew the transaction lease |
| 62 | // every x/NUM_TXN_RENEWAL_PER_REQUEST seconds. After the Core completes the request it stops renewing the |
| 63 | // transaction and sets the transaction value to TRANSACTION_COMPLETE. If the processing Core crashes then it |
| 64 | // will not renew the transaction causing the KV store to delete the transaction after its renewal period. The |
| 65 | // Core watching the transaction will then take over. |
| 66 | // Since the MIN_TXN_RENEWAL_INTERVAL_IN_SEC is 3 seconds then for any transaction that completes within 3 seconds |
| 67 | // there won't be a transaction renewal done. |
| 68 | const ( |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 69 | NumTxnRenewalPerRequest = 2 |
| 70 | MinTxnRenewalIntervalInSec = 3 |
| 71 | MinTxnReservationDurationInSec = 5 |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 72 | ) |
| 73 | |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 74 | // TransactionContext represent transaction context attributes |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 75 | type TransactionContext struct { |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 76 | kvClient kvstore.Client |
| 77 | kvOperationTimeout int |
| 78 | owner string |
| 79 | txnPrefix string |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 80 | } |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 81 | |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 82 | var ctx *TransactionContext |
| 83 | |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 84 | var txnState = []string{ |
| 85 | "UNKNOWN", |
| 86 | "SEIZED-BY-SELF", |
| 87 | "COMPLETED-BY-OTHER", |
| 88 | "ABANDONED-BY-OTHER", |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 89 | "ABANDONED_WATCH_BY_SELF"} |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 90 | |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 91 | // NewTransactionContext creates transaction context instance |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 92 | func NewTransactionContext( |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 93 | owner string, |
| 94 | txnPrefix string, |
| 95 | kvClient kvstore.Client, |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 96 | kvOpTimeout int) *TransactionContext { |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 97 | |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 98 | return &TransactionContext{ |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 99 | owner: owner, |
| 100 | txnPrefix: txnPrefix, |
| 101 | kvClient: kvClient, |
| 102 | kvOperationTimeout: kvOpTimeout} |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 103 | } |
| 104 | |
| 105 | /* |
| 106 | * Before instantiating a KVTransaction, a TransactionContext must be created. |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 107 | * The parameters stored in the context govern the behavior of all KVTransaction |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 108 | * instances. |
| 109 | * |
| 110 | * :param owner: The owner (i.e. voltha core name) of a transaction |
| 111 | * :param txnPrefix: The key prefix under which all transaction IDs, or serial numbers, |
| 112 | * will be created (e.g. "service/voltha/transactions") |
| 113 | * :param kvClient: The client API used for all interactions with the KV store. Currently |
| 114 | * only the etcd client is supported. |
Richard Jankowski | 199fd86 | 2019-03-18 14:49:51 -0400 | [diff] [blame] | 115 | * :param: kvOpTimeout: The maximum time, in seconds, to be taken by any KV operation |
| 116 | * used by this package |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 117 | */ |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 118 | |
| 119 | // SetTransactionContext creates new transaction context |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 120 | func SetTransactionContext(owner string, |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 121 | txnPrefix string, |
| 122 | kvClient kvstore.Client, |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 123 | kvOpTimeout int) error { |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 124 | |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 125 | ctx = NewTransactionContext(owner, txnPrefix, kvClient, kvOpTimeout) |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 126 | return nil |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 127 | } |
| 128 | |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 129 | // KVTransaction represent KV transaction attributes |
Richard Jankowski | e4d7766 | 2018-10-17 13:53:21 -0400 | [diff] [blame] | 130 | type KVTransaction struct { |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 131 | monitorCh chan int |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 132 | txnID string |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 133 | txnKey string |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 134 | } |
| 135 | |
| 136 | /* |
| 137 | * A KVTransaction constructor |
| 138 | * |
| 139 | * :param txnId: The serial number of a voltha request. |
Richard Jankowski | e4d7766 | 2018-10-17 13:53:21 -0400 | [diff] [blame] | 140 | * :return: A KVTransaction instance |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 141 | */ |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 142 | |
| 143 | // NewKVTransaction creates KV transaction instance |
| 144 | func NewKVTransaction(txnID string) *KVTransaction { |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 145 | return &KVTransaction{ |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 146 | txnID: txnID, |
| 147 | txnKey: ctx.txnPrefix + txnID} |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 148 | } |
| 149 | |
| 150 | /* |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 151 | * Acquired is invoked by a Core, upon reception of a request, to reserve the transaction key in the KV store. The |
| 152 | * request may be resource specific (i.e will include an ID for that resource) or may be generic (i.e. list a set of |
| 153 | * resources). If the request is resource specific then this function should be invoked with the ownedByMe flag to |
| 154 | * indicate whether this Core owns this resource. In the case where this Core owns this resource or it is a generic |
| 155 | * request then we will proceed to reserve the transaction key in the KV store for a minimum time specified by the |
| 156 | * minDuration param. If the reservation request fails (i.e. the other Core got the reservation before this one - this |
| 157 | * can happen only for generic request) then the Core will start to watch for changes to the key to determine |
| 158 | * whether the other Core completed the transaction successfully or the Core just died. If the Core does not own the |
| 159 | * resource then we will proceed to watch the transaction key. |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 160 | * |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 161 | * :param minDuration: minimum time to reserve the transaction key in the KV store |
| 162 | * :param ownedByMe: specify whether the request is about a resource owned or not. If it's absent then this is a |
| 163 | * generic request that has no specific resource ID (e.g. list) |
| 164 | * |
| 165 | * :return: A boolean specifying whether the resource was acquired. An error is return in case this function is invoked |
| 166 | * for a resource that is nonexistent. |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 167 | */ |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 168 | |
| 169 | // Acquired aquires transaction status |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 170 | func (c *KVTransaction) Acquired(ctx context.Context, minDuration int64, ownedByMe ...bool) (bool, error) { |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 171 | var acquired bool |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 172 | var currOwner string |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 173 | var res int |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 174 | |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 175 | // Convert milliseconds to seconds, rounding up |
| 176 | // The reservation TTL is specified in seconds |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 177 | durationInSecs := minDuration / 1000 |
| 178 | if remainder := minDuration % 1000; remainder > 0 { |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 179 | durationInSecs++ |
| 180 | } |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 181 | if durationInSecs < int64(MinTxnReservationDurationInSec) { |
| 182 | durationInSecs = int64(MinTxnReservationDurationInSec) |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 183 | } |
| 184 | genericRequest := true |
| 185 | resourceOwned := false |
| 186 | if len(ownedByMe) > 0 { |
| 187 | genericRequest = false |
| 188 | resourceOwned = ownedByMe[0] |
| 189 | } |
| 190 | if resourceOwned || genericRequest { |
| 191 | // Keep the reservation longer that the minDuration (which is really the request timeout) to ensure the |
| 192 | // transaction key stays in the KV store until after the Core finalize a request timeout condition (which is |
| 193 | // a success from a request completion perspective). |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 194 | if err := c.tryToReserveTxn(ctx, durationInSecs*2); err == nil { |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 195 | res = SeizedBySelf |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 196 | } else { |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 197 | logger.Debugw("watch-other-server", |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 198 | log.Fields{"transactionId": c.txnID, "owner": currOwner, "timeout": durationInSecs}) |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 199 | res = c.Watch(ctx, durationInSecs) |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 200 | } |
| 201 | } else { |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 202 | res = c.Watch(ctx, durationInSecs) |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 203 | } |
| 204 | switch res { |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 205 | case SeizedBySelf, AbandonedByOther: |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 206 | acquired = true |
| 207 | default: |
| 208 | acquired = false |
| 209 | } |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 210 | logger.Debugw("acquire-transaction-status", log.Fields{"transactionId": c.txnID, "acquired": acquired, "result": txnState[res]}) |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 211 | return acquired, nil |
| 212 | } |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 213 | |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 214 | func (c *KVTransaction) tryToReserveTxn(ctxt context.Context, durationInSecs int64) error { |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 215 | var currOwner string |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 216 | var res int |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 217 | var err error |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 218 | value, _ := ctx.kvClient.Reserve(ctxt, c.txnKey, ctx.owner, durationInSecs) |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 219 | if value != nil { |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 220 | if currOwner, err = kvstore.ToString(value); err != nil { // This should never happen |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 221 | logger.Errorw("unexpected-owner-type", log.Fields{"transactionId": c.txnID, "error": err}) |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 222 | return err |
| 223 | } |
| 224 | if currOwner == ctx.owner { |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 225 | logger.Debugw("acquired-transaction", log.Fields{"transactionId": c.txnID, "result": txnState[res]}) |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 226 | // Setup the monitoring channel |
| 227 | c.monitorCh = make(chan int) |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 228 | go c.holdOnToTxnUntilProcessingCompleted(ctxt, c.txnKey, ctx.owner, durationInSecs) |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 229 | return nil |
khenaidoo | 89b0e94 | 2018-10-21 21:11:33 -0400 | [diff] [blame] | 230 | } |
| 231 | } |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 232 | return status.Error(codes.PermissionDenied, "reservation-denied") |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 233 | } |
| 234 | |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 235 | // Watch watches transaction |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 236 | func (c *KVTransaction) Watch(ctxt context.Context, durationInSecs int64) int { |
Richard Jankowski | 199fd86 | 2019-03-18 14:49:51 -0400 | [diff] [blame] | 237 | var res int |
Scott Baker | 0e78ba2 | 2020-02-24 17:58:47 -0800 | [diff] [blame] | 238 | events := ctx.kvClient.Watch(ctxt, c.txnKey, false) |
A R Karthick | 43ba1fb | 2019-10-03 16:24:21 +0000 | [diff] [blame] | 239 | defer ctx.kvClient.CloseWatch(c.txnKey, events) |
Richard Jankowski | 199fd86 | 2019-03-18 14:49:51 -0400 | [diff] [blame] | 240 | |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 241 | transactionWasAcquiredByOther := false |
| 242 | |
| 243 | //Check whether the transaction was already completed by the other Core before we got here. |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 244 | if kvp, _ := ctx.kvClient.Get(ctxt, c.txnKey); kvp != nil { |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 245 | transactionWasAcquiredByOther = true |
| 246 | if val, err := kvstore.ToString(kvp.Value); err == nil { |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 247 | if val == TransactionComplete { |
| 248 | res = CompletedByOther |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 249 | // Do an immediate delete of the transaction in the KV Store to free up KV Storage faster |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 250 | err = c.Delete(ctxt) |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 251 | if err != nil { |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 252 | logger.Errorw("unable-to-delete-the-transaction", log.Fields{"error": err}) |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 253 | } |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 254 | return res |
| 255 | } |
| 256 | } else { |
| 257 | // An unexpected value - let's get out of here as something did not go according to plan |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 258 | res = AbandonedWatchBySelf |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 259 | logger.Debugw("cannot-read-transaction-value", log.Fields{"txn": c.txnID, "error": err}) |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 260 | return res |
| 261 | } |
| 262 | } |
| 263 | |
A R Karthick | 919f6db | 2019-08-29 18:14:56 +0000 | [diff] [blame] | 264 | for { |
| 265 | select { |
A R Karthick | 919f6db | 2019-08-29 18:14:56 +0000 | [diff] [blame] | 266 | case event := <-events: |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 267 | transactionWasAcquiredByOther = true |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 268 | logger.Debugw("received-event", log.Fields{"txn": c.txnID, "type": event.EventType}) |
A R Karthick | 919f6db | 2019-08-29 18:14:56 +0000 | [diff] [blame] | 269 | if event.EventType == kvstore.DELETE { |
| 270 | // The other core failed to process the request |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 271 | res = AbandonedByOther |
A R Karthick | 919f6db | 2019-08-29 18:14:56 +0000 | [diff] [blame] | 272 | } else if event.EventType == kvstore.PUT { |
| 273 | key, e1 := kvstore.ToString(event.Key) |
| 274 | val, e2 := kvstore.ToString(event.Value) |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 275 | if e1 == nil && e2 == nil && key == c.txnKey { |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 276 | if val == TransactionComplete { |
| 277 | res = CompletedByOther |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 278 | // Successful request completion has been detected. Remove the transaction key |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 279 | err := c.Delete(ctxt) |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 280 | if err != nil { |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 281 | logger.Errorw("unable-to-delete-the-transaction", log.Fields{"error": err}) |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 282 | } |
A R Karthick | 919f6db | 2019-08-29 18:14:56 +0000 | [diff] [blame] | 283 | } else { |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 284 | logger.Debugw("Ignoring-PUT-event", log.Fields{"val": val, "key": key}) |
A R Karthick | 919f6db | 2019-08-29 18:14:56 +0000 | [diff] [blame] | 285 | continue |
| 286 | } |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 287 | } else { |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 288 | logger.Warnw("received-unexpected-PUT-event", log.Fields{"txn": c.txnID, "key": key, "ctxKey": c.txnKey}) |
A R Karthick | 919f6db | 2019-08-29 18:14:56 +0000 | [diff] [blame] | 289 | } |
Richard Jankowski | 199fd86 | 2019-03-18 14:49:51 -0400 | [diff] [blame] | 290 | } |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 291 | case <-time.After(time.Duration(durationInSecs) * time.Second): |
| 292 | // Corner case: In the case where the Core owning the device dies and before this Core takes ownership of |
| 293 | // this device there is a window where new requests will end up being watched instead of being processed. |
| 294 | // Grab the request if the other Core did not create the transaction in the KV store. |
| 295 | // TODO: Use a peer-monitoring probe to switch over (still relies on the probe frequency). This will |
| 296 | // guarantee that the peer is actually gone instead of limiting the time the peer can get hold of a |
| 297 | // request. |
| 298 | if !transactionWasAcquiredByOther { |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 299 | logger.Debugw("timeout-no-peer", log.Fields{"txId": c.txnID}) |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 300 | res = AbandonedByOther |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 301 | } else { |
| 302 | continue |
| 303 | } |
Richard Jankowski | 199fd86 | 2019-03-18 14:49:51 -0400 | [diff] [blame] | 304 | } |
A R Karthick | 919f6db | 2019-08-29 18:14:56 +0000 | [diff] [blame] | 305 | break |
Richard Jankowski | 199fd86 | 2019-03-18 14:49:51 -0400 | [diff] [blame] | 306 | } |
| 307 | return res |
| 308 | } |
| 309 | |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 310 | // Close closes transaction |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 311 | func (c *KVTransaction) Close(ctxt context.Context) error { |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 312 | logger.Debugw("close", log.Fields{"txn": c.txnID}) |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 313 | // Stop monitoring the key (applies only when there has been no transaction switch over) |
| 314 | if c.monitorCh != nil { |
| 315 | close(c.monitorCh) |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 316 | err := ctx.kvClient.Put(ctxt, c.txnKey, TransactionComplete) |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 317 | |
| 318 | if err != nil { |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 319 | logger.Errorw("unable-to-write-a-key-value-pair-to-the-KV-store", log.Fields{"error": err}) |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 320 | } |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 321 | } |
| 322 | return nil |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 323 | } |
| 324 | |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 325 | // Delete deletes transaction |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 326 | func (c *KVTransaction) Delete(ctxt context.Context) error { |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 327 | logger.Debugw("delete", log.Fields{"txn": c.txnID}) |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 328 | return ctx.kvClient.Delete(ctxt, c.txnKey) |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 329 | } |
| 330 | |
| 331 | // holdOnToTxnUntilProcessingCompleted renews the transaction lease until the transaction is complete. durationInSecs |
| 332 | // is used to calculate the frequency at which the Core processing the transaction renews the lease. This function |
| 333 | // exits only when the transaction is Closed, i.e completed. |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 334 | func (c *KVTransaction) holdOnToTxnUntilProcessingCompleted(ctxt context.Context, key string, owner string, durationInSecs int64) { |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 335 | logger.Debugw("holdOnToTxnUntilProcessingCompleted", log.Fields{"txn": c.txnID}) |
npujar | 1d86a52 | 2019-11-14 17:11:16 +0530 | [diff] [blame] | 336 | renewInterval := durationInSecs / NumTxnRenewalPerRequest |
| 337 | if renewInterval < MinTxnRenewalIntervalInSec { |
| 338 | renewInterval = MinTxnRenewalIntervalInSec |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 339 | } |
| 340 | forLoop: |
| 341 | for { |
| 342 | select { |
| 343 | case <-c.monitorCh: |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 344 | logger.Debugw("transaction-renewal-exits", log.Fields{"txn": c.txnID}) |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 345 | break forLoop |
| 346 | case <-time.After(time.Duration(renewInterval) * time.Second): |
npujar | 467fe75 | 2020-01-16 20:17:45 +0530 | [diff] [blame] | 347 | if err := ctx.kvClient.RenewReservation(ctxt, c.txnKey); err != nil { |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 348 | // Log and continue. |
Girish Kumar | f56a468 | 2020-03-20 20:07:46 +0000 | [diff] [blame^] | 349 | logger.Warnw("transaction-renewal-failed", log.Fields{"txnId": c.txnKey, "error": err}) |
khenaidoo | 09771ef | 2019-10-11 14:25:02 -0400 | [diff] [blame] | 350 | } |
| 351 | } |
| 352 | } |
Richard Jankowski | 215a3e2 | 2018-10-04 13:56:11 -0400 | [diff] [blame] | 353 | } |