oceanbase

1
/**
2
 * Copyright (c) 2021 OceanBase
3
 * OceanBase CE is licensed under Mulan PubL v2.
4
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
5
 * You may obtain a copy of Mulan PubL v2 at:
6
 *          http://license.coscl.org.cn/MulanPubL-2.0
7
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
8
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
9
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
10
 * See the Mulan PubL v2 for more details.
11
 */
12

13
#define USING_LOG_PREFIX SERVER_OMT
14
#include "observer/omt/ob_multi_tenant.h"
15
#include "observer/omt/ob_multi_tenant_operator.h"
16
#include "share/rc/ob_tenant_base.h"
17
#include "observer/ob_server_struct.h"
18
#include "observer/omt/ob_tenant.h"
19

20
namespace oceanbase
21
{
22
namespace omt
23
{
24

25
ObMultiTenantOperator::ObMultiTenantOperator() :
26
  inited_(false),
27
  tenant_idx_(-1),
28
  tenant_(nullptr)
29
{}
30

31
ObMultiTenantOperator::~ObMultiTenantOperator()
32
{
33
  reset();
34
}
35

36
void ObMultiTenantOperator::reset()
37
{
38
  if (inited_) {
39
    // 如果迭代未结束，释放残留的资源
40
    int ret = OB_SUCCESS;
41
    if (tenant_idx_ >= 0 && tenant_idx_ < tenant_ids_.size()) {
42
      uint64_t tenant_id = tenant_ids_.at(tenant_idx_);
43
      if (tenant_ != nullptr) {
44
        if (tenant_id != tenant_->id()) {
45
          LOG_ERROR("ObMultiTenantOperator::reset", K(tenant_id), K(tenant_->id()));
46
          abort();
47
        }
48
        {
49
          share::ObTenantSwitchGuard guard(tenant_);
50
          release_last_tenant();
51
        }
52
        tenant_->unlock(handle_);
53
      }
54
    }
55
    if (OB_FAIL(ret)) {
56
      // 无法处理资源释放不了的问题
57
      abort();
58
    } else {
59
      tenant_ids_.reuse();
60
      tenant_idx_ = -1;
61
      tenant_ = nullptr;
62
      inited_ = false;
63
    }
64
  }
65
}
66

67
int ObMultiTenantOperator::init()
68
{
69
  int ret = OB_SUCCESS;
70
  ObMultiTenant *omt = GCTX.omt_;
71
  TenantIdList tenant_list;
72
  if (omt == nullptr) {
73
    ret = OB_ERR_UNEXPECTED;
74
    LOG_ERROR("operator init", K(ret));
75
  } else if (inited_) {
76
    ret = OB_INIT_TWICE;
77
    LOG_ERROR("operator init", K(ret));
78
  } else {
79
    if (tenant_ids_.size() != 0 || tenant_idx_ != -1) {
80
      ret = OB_ERR_UNEXPECTED;
81
      LOG_ERROR("operator init", K(ret), K(tenant_idx_), K(tenant_ids_));
82
    } else if (FALSE_IT(omt->get_tenant_ids(tenant_list))) {
83
    } else {
84
      for (int i=0; i < tenant_list.size(); i++) {
85
        uint64_t tenant_id = tenant_list.at(i);
86
        if (!is_virtual_tenant_id(tenant_id) && is_need_process(tenant_id)) {
87
          tenant_ids_.push_back(tenant_id);
88
        }
89
      }
90
      inited_ = true;
91
    }
92
  }
93
  return ret;
94
}
95

96
int ObMultiTenantOperator::execute(common::ObNewRow *&row)
97
{
98
  int ret = OB_SUCCESS;
99
  if (!inited_) {
100
    ret = init();
101
  }
102
  if (OB_SUCC(ret)) {
103
    if (tenant_idx_ == -1) {
104
      tenant_idx_ = 0;
105
    }
106
    while (OB_SUCC(ret)) {
107
      if (tenant_idx_ >= tenant_ids_.size()) {
108
        // finish iter
109
        ret = OB_ITER_END;
110
      } else {
111
        uint64_t tenant_id = tenant_ids_.at(tenant_idx_);
112
        int process_ret = OB_SUCCESS;
113
        if (tenant_ == nullptr) {
114
          if (OB_FAIL(GCTX.omt_->get_active_tenant_with_tenant_lock(tenant_id, handle_, tenant_))) {
115
            LOG_WARN("get_tenant_with_tenant_lock", K(ret), K(tenant_id));
116
          }
117
        } else {
118
          // check iter tenant
119
          if (tenant_->id() != tenant_id) {
120
            LOG_ERROR("ObMultiTenantOperator tenant mismatch", K(tenant_ids_), K(tenant_idx_), K(tenant_id), K(tenant_->id()));
121
            abort();
122
          }
123
        }
124
        if (OB_SUCC(ret)) {
125
          share::ObTenantSwitchGuard guard(tenant_);
126
          process_ret = process_curr_tenant(row);
127
        }
128
        if (OB_SUCC(ret)) {
129
          if (process_ret == OB_SUCCESS) {
130
            // succ do nothing
131
            break;
132
          } else if (process_ret == OB_ITER_END) {
133
            {
134
              // release last tenant obj
135
              share::ObTenantSwitchGuard guard(tenant_);
136
              release_last_tenant();
137
            }
138
            tenant_->unlock(handle_);
139
            tenant_ = nullptr;
140
            tenant_idx_++;
141
          } else {
142
            ret = process_ret;
143
            LOG_WARN("operator process", K(ret), K(lbt()));
144
          }
145
        } else if (ret == OB_TENANT_NOT_IN_SERVER) {
146
          ret = OB_SUCCESS;
147
          tenant_idx_++;
148
        }
149
      }
150
    }
151
  }
152
  return ret;
153
}
154

155
} // end omt
156
} // end oceanbase
157