You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

101 lines
2.4 KiB

  1. package hbase
  2. import (
  3. pb "github.com/golang/protobuf/proto"
  4. "github.com/juju/errors"
  5. "github.com/ngaut/log"
  6. "github.com/pingcap/go-hbase/proto"
  7. )
  8. type action interface {
  9. ToProto() pb.Message
  10. }
  11. func (c *client) innerCall(table, row []byte, action action, useCache bool) (*call, error) {
  12. region, err := c.LocateRegion(table, row, useCache)
  13. if err != nil {
  14. return nil, errors.Trace(err)
  15. }
  16. conn, err := c.getClientConn(region.Server)
  17. if err != nil {
  18. return nil, errors.Trace(err)
  19. }
  20. regionSpecifier := &proto.RegionSpecifier{
  21. Type: proto.RegionSpecifier_REGION_NAME.Enum(),
  22. Value: []byte(region.Name),
  23. }
  24. var cl *call
  25. switch a := action.(type) {
  26. case *Get:
  27. cl = newCall(&proto.GetRequest{
  28. Region: regionSpecifier,
  29. Get: a.ToProto().(*proto.Get),
  30. })
  31. case *Put, *Delete:
  32. cl = newCall(&proto.MutateRequest{
  33. Region: regionSpecifier,
  34. Mutation: a.ToProto().(*proto.MutationProto),
  35. })
  36. case *CoprocessorServiceCall:
  37. cl = newCall(&proto.CoprocessorServiceRequest{
  38. Region: regionSpecifier,
  39. Call: a.ToProto().(*proto.CoprocessorServiceCall),
  40. })
  41. default:
  42. return nil, errors.Errorf("Unknown action - %T - %v", action, action)
  43. }
  44. err = conn.call(cl)
  45. if err != nil {
  46. // If failed, remove bad server conn cache.
  47. cachedKey := cachedConnKey(region.Server, ClientService)
  48. delete(c.cachedConns, cachedKey)
  49. return nil, errors.Trace(err)
  50. }
  51. return cl, nil
  52. }
  53. func (c *client) innerDo(table, row []byte, action action, useCache bool) (pb.Message, error) {
  54. // Try to create and send a new resuqest call.
  55. cl, err := c.innerCall(table, row, action, useCache)
  56. if err != nil {
  57. log.Warnf("inner call failed - %v", errors.ErrorStack(err))
  58. return nil, errors.Trace(err)
  59. }
  60. // Wait and receive the result.
  61. return <-cl.responseCh, nil
  62. }
  63. func (c *client) do(table, row []byte, action action, useCache bool) (pb.Message, error) {
  64. var (
  65. result pb.Message
  66. err error
  67. )
  68. LOOP:
  69. for i := 0; i < c.maxRetries; i++ {
  70. result, err = c.innerDo(table, row, action, useCache)
  71. if err == nil {
  72. switch r := result.(type) {
  73. case *exception:
  74. err = errors.New(r.msg)
  75. // If get an execption response, clean old region cache.
  76. c.CleanRegionCache(table)
  77. default:
  78. break LOOP
  79. }
  80. }
  81. useCache = false
  82. log.Warnf("Retrying action for the %d time(s), error - %v", i+1, errors.ErrorStack(err))
  83. retrySleep(i + 1)
  84. }
  85. return result, errors.Trace(err)
  86. }