diff --git a/accounts/abi/bind/backends/simulated.go b/accounts/abi/bind/backends/simulated.go index 0b2555c994..f7f3dec839 100644 --- a/accounts/abi/bind/backends/simulated.go +++ b/accounts/abi/bind/backends/simulated.go @@ -127,15 +127,28 @@ func (b *SimulatedBackend) rollback() { b.pendingState, _ = state.New(b.pendingBlock.Root(), statedb.Database()) } +// stateByBlockNumber retrieves a state by a given blocknumber. +func (b *SimulatedBackend) stateByBlockNumber(ctx context.Context, blockNumber *big.Int) (*state.StateDB, error) { + if blockNumber == nil || blockNumber.Cmp(b.blockchain.CurrentBlock().Number()) == 0 { + return b.blockchain.State() + } + block, err := b.BlockByNumber(ctx, blockNumber) + if err != nil { + return nil, err + } + return b.blockchain.StateAt(block.Hash()) +} + // CodeAt returns the code associated with a certain account in the blockchain. func (b *SimulatedBackend) CodeAt(ctx context.Context, contract common.Address, blockNumber *big.Int) ([]byte, error) { b.mu.Lock() defer b.mu.Unlock() - if blockNumber != nil && blockNumber.Cmp(b.blockchain.CurrentBlock().Number()) != 0 { - return nil, errBlockNumberUnsupported + statedb, err := b.stateByBlockNumber(ctx, blockNumber) + if err != nil { + return nil, err } - statedb, _ := b.blockchain.State() + return statedb.GetCode(contract), nil } @@ -144,10 +157,11 @@ func (b *SimulatedBackend) BalanceAt(ctx context.Context, contract common.Addres b.mu.Lock() defer b.mu.Unlock() - if blockNumber != nil && blockNumber.Cmp(b.blockchain.CurrentBlock().Number()) != 0 { - return nil, errBlockNumberUnsupported + statedb, err := b.stateByBlockNumber(ctx, blockNumber) + if err != nil { + return nil, err } - statedb, _ := b.blockchain.State() + return statedb.GetBalance(contract), nil } @@ -156,10 +170,11 @@ func (b *SimulatedBackend) NonceAt(ctx context.Context, contract common.Address, b.mu.Lock() defer b.mu.Unlock() - if blockNumber != nil && blockNumber.Cmp(b.blockchain.CurrentBlock().Number()) != 0 { - return 0, errBlockNumberUnsupported + statedb, err := b.stateByBlockNumber(ctx, blockNumber) + if err != nil { + return 0, err } - statedb, _ := b.blockchain.State() + return statedb.GetNonce(contract), nil } @@ -168,10 +183,11 @@ func (b *SimulatedBackend) StorageAt(ctx context.Context, contract common.Addres b.mu.Lock() defer b.mu.Unlock() - if blockNumber != nil && blockNumber.Cmp(b.blockchain.CurrentBlock().Number()) != 0 { - return nil, errBlockNumberUnsupported + statedb, err := b.stateByBlockNumber(ctx, blockNumber) + if err != nil { + return nil, err } - statedb, _ := b.blockchain.State() + val := statedb.GetState(contract, key) return val[:], nil } diff --git a/accounts/manager.go b/accounts/manager.go index 731d12ea30..acf41ed8e9 100644 --- a/accounts/manager.go +++ b/accounts/manager.go @@ -141,6 +141,11 @@ func (am *Manager) Wallets() []Wallet { am.lock.RLock() defer am.lock.RUnlock() + return am.walletsNoLock() +} + +// walletsNoLock returns all registered wallets. Callers must hold am.lock. +func (am *Manager) walletsNoLock() []Wallet { cpy := make([]Wallet, len(am.wallets)) copy(cpy, am.wallets) return cpy @@ -155,7 +160,7 @@ func (am *Manager) Wallet(url string) (Wallet, error) { if err != nil { return nil, err } - for _, wallet := range am.Wallets() { + for _, wallet := range am.walletsNoLock() { if wallet.URL() == parsed { return wallet, nil } diff --git a/cmd/geth/retesteth.go b/cmd/geth/retesteth.go index 28e792bd77..ab7981e518 100644 --- a/cmd/geth/retesteth.go +++ b/cmd/geth/retesteth.go @@ -80,6 +80,7 @@ type RetestethEthAPI interface { SendRawTransaction(ctx context.Context, rawTx hexutil.Bytes) (common.Hash, error) BlockNumber(ctx context.Context) (uint64, error) GetBlockByNumber(ctx context.Context, blockNr math.HexOrDecimal64, fullTx bool) (map[string]interface{}, error) + GetBlockByHash(ctx context.Context, blockHash common.Hash, fullTx bool) (map[string]interface{}, error) GetBalance(ctx context.Context, address common.Address, blockNr math.HexOrDecimal64) (*math.HexOrDecimal256, error) GetCode(ctx context.Context, address common.Address, blockNr math.HexOrDecimal64) (hexutil.Bytes, error) GetTransactionCount(ctx context.Context, address common.Address, blockNr math.HexOrDecimal64) (uint64, error) @@ -618,6 +619,20 @@ func (api *RetestethAPI) GetBlockByNumber(ctx context.Context, blockNr math.HexO return nil, fmt.Errorf("block %d not found", blockNr) } +func (api *RetestethAPI) GetBlockByHash(ctx context.Context, blockHash common.Hash, fullTx bool) (map[string]interface{}, error) { + block := api.blockchain.GetBlockByHash(blockHash) + if block != nil { + response, err := RPCMarshalBlock(block, true, fullTx) + if err != nil { + return nil, err + } + response["author"] = response["miner"] + response["totalDifficulty"] = (*hexutil.Big)(api.blockchain.GetTd(block.Hash(), block.Number().Uint64())) + return response, err + } + return nil, fmt.Errorf("block 0x%x not found", blockHash) +} + func (api *RetestethAPI) AccountRange(ctx context.Context, blockHashOrNumber *math.HexOrDecimal256, txIndex uint64, addressHash *math.HexOrDecimal256, maxResults uint64, diff --git a/common/mclock/mclock.go b/common/mclock/mclock.go index d0e0cd78be..3aca257cb3 100644 --- a/common/mclock/mclock.go +++ b/common/mclock/mclock.go @@ -31,44 +31,93 @@ func Now() AbsTime { return AbsTime(monotime.Now()) } -// Add returns t + d. +// Add returns t + d as absolute time. func (t AbsTime) Add(d time.Duration) AbsTime { return t + AbsTime(d) } +// Sub returns t - t2 as a duration. +func (t AbsTime) Sub(t2 AbsTime) time.Duration { + return time.Duration(t - t2) +} + // The Clock interface makes it possible to replace the monotonic system clock with // a simulated clock. type Clock interface { Now() AbsTime Sleep(time.Duration) - After(time.Duration) <-chan time.Time + NewTimer(time.Duration) ChanTimer + After(time.Duration) <-chan AbsTime AfterFunc(d time.Duration, f func()) Timer } -// Timer represents a cancellable event returned by AfterFunc +// Timer is a cancellable event created by AfterFunc. type Timer interface { + // Stop cancels the timer. It returns false if the timer has already + // expired or been stopped. Stop() bool } +// ChanTimer is a cancellable event created by NewTimer. +type ChanTimer interface { + Timer + + // The channel returned by C receives a value when the timer expires. + C() <-chan AbsTime + // Reset reschedules the timer with a new timeout. + // It should be invoked only on stopped or expired timers with drained channels. + Reset(time.Duration) +} + // System implements Clock using the system clock. type System struct{} // Now returns the current monotonic time. -func (System) Now() AbsTime { +func (c System) Now() AbsTime { return AbsTime(monotime.Now()) } // Sleep blocks for the given duration. -func (System) Sleep(d time.Duration) { +func (c System) Sleep(d time.Duration) { time.Sleep(d) } +// NewTimer creates a timer which can be rescheduled. +func (c System) NewTimer(d time.Duration) ChanTimer { + ch := make(chan AbsTime, 1) + t := time.AfterFunc(d, func() { + // This send is non-blocking because that's how time.Timer + // behaves. It doesn't matter in the happy case, but does + // when Reset is misused. + select { + case ch <- c.Now(): + default: + } + }) + return &systemTimer{t, ch} +} + // After returns a channel which receives the current time after d has elapsed. -func (System) After(d time.Duration) <-chan time.Time { - return time.After(d) +func (c System) After(d time.Duration) <-chan AbsTime { + ch := make(chan AbsTime, 1) + time.AfterFunc(d, func() { ch <- c.Now() }) + return ch } // AfterFunc runs f on a new goroutine after the duration has elapsed. -func (System) AfterFunc(d time.Duration, f func()) Timer { +func (c System) AfterFunc(d time.Duration, f func()) Timer { return time.AfterFunc(d, f) } + +type systemTimer struct { + *time.Timer + ch <-chan AbsTime +} + +func (st *systemTimer) Reset(d time.Duration) { + st.Timer.Reset(d) +} + +func (st *systemTimer) C() <-chan AbsTime { + return st.ch +} diff --git a/common/mclock/simclock.go b/common/mclock/simclock.go index 4d351252ff..766ca0f873 100644 --- a/common/mclock/simclock.go +++ b/common/mclock/simclock.go @@ -17,6 +17,7 @@ package mclock import ( + "container/heap" "sync" "time" ) @@ -32,18 +33,24 @@ import ( // the timeout using a channel or semaphore. type Simulated struct { now AbsTime - scheduled []*simTimer + scheduled simTimerHeap mu sync.RWMutex cond *sync.Cond - lastId uint64 } -// simTimer implements Timer on the virtual clock. +// simTimer implements ChanTimer on the virtual clock. type simTimer struct { - do func() - at AbsTime - id uint64 - s *Simulated + at AbsTime + index int // position in s.scheduled + s *Simulated + do func() + ch <-chan AbsTime +} + +func (s *Simulated) init() { + if s.cond == nil { + s.cond = sync.NewCond(&s.mu) + } } // Run moves the clock by the given duration, executing all timers before that duration. @@ -53,14 +60,9 @@ func (s *Simulated) Run(d time.Duration) { end := s.now + AbsTime(d) var do []func() - for len(s.scheduled) > 0 { - ev := s.scheduled[0] - if ev.at > end { - break - } - s.now = ev.at + for len(s.scheduled) > 0 && s.scheduled[0].at <= end { + ev := heap.Pop(&s.scheduled).(*simTimer) do = append(do, ev.do) - s.scheduled = s.scheduled[1:] } s.now = end s.mu.Unlock() @@ -102,14 +104,22 @@ func (s *Simulated) Sleep(d time.Duration) { <-s.After(d) } +// NewTimer creates a timer which fires when the clock has advanced by d. +func (s *Simulated) NewTimer(d time.Duration) ChanTimer { + s.mu.Lock() + defer s.mu.Unlock() + + ch := make(chan AbsTime, 1) + var timer *simTimer + timer = s.schedule(d, func() { ch <- timer.at }) + timer.ch = ch + return timer +} + // After returns a channel which receives the current time after the clock // has advanced by d. -func (s *Simulated) After(d time.Duration) <-chan time.Time { - after := make(chan time.Time, 1) - s.AfterFunc(d, func() { - after <- (time.Time{}).Add(time.Duration(s.now)) - }) - return after +func (s *Simulated) After(d time.Duration) <-chan AbsTime { + return s.NewTimer(d).C() } // AfterFunc runs fn after the clock has advanced by d. Unlike with the system @@ -117,46 +127,83 @@ func (s *Simulated) After(d time.Duration) <-chan time.Time { func (s *Simulated) AfterFunc(d time.Duration, fn func()) Timer { s.mu.Lock() defer s.mu.Unlock() + + return s.schedule(d, fn) +} + +func (s *Simulated) schedule(d time.Duration, fn func()) *simTimer { s.init() at := s.now + AbsTime(d) - s.lastId++ - id := s.lastId - l, h := 0, len(s.scheduled) - ll := h - for l != h { - m := (l + h) / 2 - if (at < s.scheduled[m].at) || ((at == s.scheduled[m].at) && (id < s.scheduled[m].id)) { - h = m - } else { - l = m + 1 - } - } ev := &simTimer{do: fn, at: at, s: s} - s.scheduled = append(s.scheduled, nil) - copy(s.scheduled[l+1:], s.scheduled[l:ll]) - s.scheduled[l] = ev + heap.Push(&s.scheduled, ev) s.cond.Broadcast() return ev } func (ev *simTimer) Stop() bool { - s := ev.s - s.mu.Lock() - defer s.mu.Unlock() + ev.s.mu.Lock() + defer ev.s.mu.Unlock() - for i := 0; i < len(s.scheduled); i++ { - if s.scheduled[i] == ev { - s.scheduled = append(s.scheduled[:i], s.scheduled[i+1:]...) - s.cond.Broadcast() - return true - } + if ev.index < 0 { + return false } - return false + heap.Remove(&ev.s.scheduled, ev.index) + ev.s.cond.Broadcast() + ev.index = -1 + return true } -func (s *Simulated) init() { - if s.cond == nil { - s.cond = sync.NewCond(&s.mu) +func (ev *simTimer) Reset(d time.Duration) { + if ev.ch == nil { + panic("mclock: Reset() on timer created by AfterFunc") } + + ev.s.mu.Lock() + defer ev.s.mu.Unlock() + ev.at = ev.s.now.Add(d) + if ev.index < 0 { + heap.Push(&ev.s.scheduled, ev) // already expired + } else { + heap.Fix(&ev.s.scheduled, ev.index) // hasn't fired yet, reschedule + } + ev.s.cond.Broadcast() +} + +func (ev *simTimer) C() <-chan AbsTime { + if ev.ch == nil { + panic("mclock: C() on timer created by AfterFunc") + } + return ev.ch +} + +type simTimerHeap []*simTimer + +func (h *simTimerHeap) Len() int { + return len(*h) +} + +func (h *simTimerHeap) Less(i, j int) bool { + return (*h)[i].at < (*h)[j].at +} + +func (h *simTimerHeap) Swap(i, j int) { + (*h)[i], (*h)[j] = (*h)[j], (*h)[i] + (*h)[i].index = i + (*h)[j].index = j +} + +func (h *simTimerHeap) Push(x interface{}) { + t := x.(*simTimer) + t.index = len(*h) + *h = append(*h, t) +} + +func (h *simTimerHeap) Pop() interface{} { + end := len(*h) - 1 + t := (*h)[end] + t.index = -1 + (*h)[end] = nil + *h = (*h)[:end] + return t } diff --git a/common/mclock/simclock_test.go b/common/mclock/simclock_test.go index 09e4391c1c..94aa4f2b39 100644 --- a/common/mclock/simclock_test.go +++ b/common/mclock/simclock_test.go @@ -25,14 +25,16 @@ var _ Clock = System{} var _ Clock = new(Simulated) func TestSimulatedAfter(t *testing.T) { - const timeout = 30 * time.Minute - const adv = time.Minute - var ( - c Simulated - end = c.Now().Add(timeout) - ch = c.After(timeout) + timeout = 30 * time.Minute + offset = 99 * time.Hour + adv = 11 * time.Minute + c Simulated ) + c.Run(offset) + + end := c.Now().Add(timeout) + ch := c.After(timeout) for c.Now() < end.Add(-adv) { c.Run(adv) select { @@ -45,8 +47,8 @@ func TestSimulatedAfter(t *testing.T) { c.Run(adv) select { case stamp := <-ch: - want := time.Time{}.Add(timeout) - if !stamp.Equal(want) { + want := AbsTime(0).Add(offset).Add(timeout) + if stamp != want { t.Errorf("Wrong time sent on timer channel: got %v, want %v", stamp, want) } default: @@ -113,3 +115,48 @@ func TestSimulatedSleep(t *testing.T) { t.Fatal("Sleep didn't return in time") } } + +func TestSimulatedTimerReset(t *testing.T) { + var ( + c Simulated + timeout = 1 * time.Hour + ) + timer := c.NewTimer(timeout) + c.Run(2 * timeout) + select { + case ftime := <-timer.C(): + if ftime != AbsTime(timeout) { + t.Fatalf("wrong time %v sent on timer channel, want %v", ftime, AbsTime(timeout)) + } + default: + t.Fatal("timer didn't fire") + } + + timer.Reset(timeout) + c.Run(2 * timeout) + select { + case ftime := <-timer.C(): + if ftime != AbsTime(3*timeout) { + t.Fatalf("wrong time %v sent on timer channel, want %v", ftime, AbsTime(3*timeout)) + } + default: + t.Fatal("timer didn't fire again") + } +} + +func TestSimulatedTimerStop(t *testing.T) { + var ( + c Simulated + timeout = 1 * time.Hour + ) + timer := c.NewTimer(timeout) + c.Run(2 * timeout) + if timer.Stop() { + t.Errorf("Stop returned true for fired timer") + } + select { + case <-timer.C(): + default: + t.Fatal("timer didn't fire") + } +} diff --git a/event/feed.go b/event/feed.go index 02f3ca6875..33dafe5886 100644 --- a/event/feed.go +++ b/event/feed.go @@ -138,6 +138,7 @@ func (f *Feed) Send(value interface{}) (nsent int) { if !f.typecheck(rvalue.Type()) { f.sendLock <- struct{}{} + f.mu.Unlock() panic(feedTypeError{op: "Send", got: rvalue.Type(), want: f.etype}) } f.mu.Unlock() diff --git a/event/subscription.go b/event/subscription.go index d03f465075..c80d171f3a 100644 --- a/event/subscription.go +++ b/event/subscription.go @@ -145,7 +145,6 @@ func (s *resubscribeSub) loop() { func (s *resubscribeSub) subscribe() Subscription { subscribed := make(chan error) var sub Subscription -retry: for { s.lastTry = mclock.Now() ctx, cancel := context.WithCancel(context.Background()) @@ -157,19 +156,19 @@ retry: select { case err := <-subscribed: cancel() - if err != nil { - // Subscribing failed, wait before launching the next try. - if s.backoffWait() { - return nil + if err == nil { + if sub == nil { + panic("event: ResubscribeFunc returned nil subscription and no error") } - continue retry + return sub } - if sub == nil { - panic("event: ResubscribeFunc returned nil subscription and no error") + // Subscribing failed, wait before launching the next try. + if s.backoffWait() { + return nil // unsubscribed during wait } - return sub case <-s.unsub: cancel() + <-subscribed // avoid leaking the s.fn goroutine. return nil } } diff --git a/event/subscription_test.go b/event/subscription_test.go index 5b8a2c8ede..c48be3aa30 100644 --- a/event/subscription_test.go +++ b/event/subscription_test.go @@ -102,7 +102,7 @@ func TestResubscribe(t *testing.T) { func TestResubscribeAbort(t *testing.T) { t.Parallel() - done := make(chan error) + done := make(chan error, 1) sub := Resubscribe(0, func(ctx context.Context) (Subscription, error) { select { case <-ctx.Done(): diff --git a/les/server_handler.go b/les/server_handler.go index 38aeac4e88..98e9294b07 100644 --- a/les/server_handler.go +++ b/les/server_handler.go @@ -128,6 +128,7 @@ func (h *serverHandler) handle(p *peer) error { } // Reject light clients if server is not synced. if !h.synced() { + p.Log().Debug("Light server not synced, rejecting peer") return p2p.DiscRequested } defer p.fcClient.Disconnect() diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go index 7c6ec94621..18ec9c69b8 100644 --- a/p2p/simulations/adapters/exec.go +++ b/p2p/simulations/adapters/exec.go @@ -287,7 +287,7 @@ func (n *ExecNode) Stop() error { if err := n.Cmd.Process.Signal(syscall.SIGTERM); err != nil { return n.Cmd.Process.Kill() } - waitErr := make(chan error) + waitErr := make(chan error, 1) go func() { waitErr <- n.Cmd.Wait() }() diff --git a/rpc/client.go b/rpc/client.go index dd06b9469e..984f56bc69 100644 --- a/rpc/client.go +++ b/rpc/client.go @@ -276,6 +276,9 @@ func (c *Client) Call(result interface{}, method string, args ...interface{}) er // The result must be a pointer so that package json can unmarshal into it. You // can also pass nil, in which case the result is ignored. func (c *Client) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error { + if result != nil && reflect.TypeOf(result).Kind() != reflect.Ptr { + return fmt.Errorf("call result parameter must be pointer or nil interface: %v", result) + } msg, err := c.newMessage(method, args...) if err != nil { return err diff --git a/rpc/client_test.go b/rpc/client_test.go index 0d55140222..9c7f72053b 100644 --- a/rpc/client_test.go +++ b/rpc/client_test.go @@ -49,6 +49,23 @@ func TestClientRequest(t *testing.T) { } } +func TestClientResponseType(t *testing.T) { + server := newTestServer() + defer server.Stop() + client := DialInProc(server) + defer client.Close() + + if err := client.Call(nil, "test_echo", "hello", 10, &echoArgs{"world"}); err != nil { + t.Errorf("Passing nil as result should be fine, but got an error: %v", err) + } + var resultVar echoResult + // Note: passing the var, not a ref + err := client.Call(resultVar, "test_echo", "hello", 10, &echoArgs{"world"}) + if err == nil { + t.Error("Passing a var as result should be an error") + } +} + func TestClientBatchRequest(t *testing.T) { server := newTestServer() defer server.Stop() diff --git a/signer/core/signed_data.go b/signer/core/signed_data.go index f512be7cea..3a827afa2d 100644 --- a/signer/core/signed_data.go +++ b/signer/core/signed_data.go @@ -923,7 +923,9 @@ func isPrimitiveTypeValid(primitiveType string) bool { primitiveType == "bytes30" || primitiveType == "bytes30[]" || primitiveType == "bytes31" || - primitiveType == "bytes31[]" { + primitiveType == "bytes31[]" || + primitiveType == "bytes32" || + primitiveType == "bytes32[]" { return true } if primitiveType == "int" ||