XRootD
XrdThrottleManager Class Reference

#include <XrdThrottleManager.hh>

+ Collaboration diagram for XrdThrottleManager:

Public Member Functions

 XrdThrottleManager (XrdSysError *lP, XrdOucTrace *tP)
 
 ~XrdThrottleManager ()
 
void Apply (int reqsize, int reqops, int uid)
 
bool CheckLoadShed (const std::string &opaque)
 
bool CloseFile (const std::string &entity)
 
void FromConfig (XrdThrottle::Configuration &config)
 
std::tuple< std::string, uint16_t > GetUserInfo (const XrdSecEntity *client)
 
unsigned long GetUserMaxConn (const std::string &username)
 
void Init ()
 
bool IsThrottling ()
 
int LoadUserLimits (const std::string &config_file)
 
bool OpenFile (const std::string &entity, std::string &open_error_message)
 
void PerformLoadShed (const std::string &opaque, std::string &host, unsigned &port)
 
void PrepLoadShed (const char *opaque, std::string &lsOpaque)
 
int ReloadUserLimits ()
 
void SetLoadShed (std::string &hostname, unsigned port, unsigned frequency)
 
void SetMaxConns (unsigned long max_conns)
 
void SetMaxOpen (unsigned long max_open)
 
void SetMaxWait (unsigned long max_wait)
 
void SetMonitor (XrdXrootdGStream *gstream)
 
void SetThrottles (float reqbyterate, float reqoprate, int concurrency, float interval_length)
 
XrdThrottleTimer StartIOTimer (uint16_t uid, bool &ok)
 

Protected Member Functions

void StopIOTimer (std::chrono::steady_clock::duration &event_duration, uint16_t uid)
 

Friends

class XrdThrottleTimer
 

Detailed Description

Definition at line 54 of file XrdThrottleManager.hh.

Constructor & Destructor Documentation

◆ XrdThrottleManager()

XrdThrottleManager::XrdThrottleManager ( XrdSysError *  lP,
XrdOucTrace *  tP 
)

Definition at line 46 of file XrdThrottleManager.cc.

46  :
47  m_trace(tP),
48  m_log(lP),
49  m_interval_length_seconds(1.0),
50  m_bytes_per_second(-1),
51  m_ops_per_second(-1),
52  m_concurrency_limit(-1),
53  m_last_round_allocation(100*1024),
54  m_loadshed_host(""),
55  m_loadshed_port(0),
56  m_loadshed_frequency(0)
57 {
58  // Until the first recompute fills them in, no user is in the wake order.
59  std::fill(m_wake_order_0.begin(), m_wake_order_0.end(), -1);
60  std::fill(m_wake_order_1.begin(), m_wake_order_1.end(), -1);
61 }

◆ ~XrdThrottleManager()

XrdThrottleManager::~XrdThrottleManager ( )
inline

Definition at line 119 of file XrdThrottleManager.hh.

119 {} // The buffmanager is never deleted

Member Function Documentation

◆ Apply()

void XrdThrottleManager::Apply ( int  reqsize,
int  reqops,
int  uid 
)

Definition at line 345 of file XrdThrottleManager.cc.

346 {
347  if (m_bytes_per_second < 0)
348  reqsize = 0;
349  if (m_ops_per_second < 0)
350  reqops = 0;
351  while (reqsize || reqops)
352  {
353  // Subtract the requested out of the shares
354  AtomicBeg(m_compute_var);
355  GetShares(m_primary_bytes_shares[uid], reqsize);
356  if (reqsize)
357  {
358  TRACE(BANDWIDTH, "Using secondary shares; request has " << reqsize << " bytes left.");
359  GetShares(m_secondary_bytes_shares[uid], reqsize);
360  TRACE(BANDWIDTH, "Finished with secondary shares; request has " << reqsize << " bytes left.");
361  }
362  else
363  {
364  TRACE(BANDWIDTH, "Filled byte shares out of primary; " << m_primary_bytes_shares[uid] << " left.");
365  }
366  GetShares(m_primary_ops_shares[uid], reqops);
367  if (reqops)
368  {
369  GetShares(m_secondary_ops_shares[uid], reqops);
370  }
371  StealShares(uid, reqsize, reqops);
372  AtomicEnd(m_compute_var);
373 
374  if (reqsize || reqops)
375  {
376  if (reqsize) TRACE(BANDWIDTH, "Sleeping to wait for throttle fairshare.");
377  if (reqops) TRACE(IOPS, "Sleeping to wait for throttle fairshare.");
378  m_compute_var.Wait();
379  m_loadshed_limit_hit++;
380  }
381  }
382 
383 }
#define AtomicBeg(Mtx)
#define AtomicEnd(Mtx)
#define TRACE(act, x)
Definition: XrdTrace.hh:63

References AtomicBeg, AtomicEnd, TRACE, and XrdSysCondVar::Wait().

+ Here is the call graph for this function:

◆ CheckLoadShed()

bool XrdThrottleManager::CheckLoadShed ( const std::string &  opaque)

Definition at line 797 of file XrdThrottleManager.cc.

798 {
799  if (m_loadshed_port == 0)
800  {
801  return false;
802  }
803  if (m_loadshed_limit_hit == 0)
804  {
805  return false;
806  }
807  if (static_cast<unsigned>(rand()) % 100 > m_loadshed_frequency)
808  {
809  return false;
810  }
811  if (opaque.empty())
812  {
813  return false;
814  }
815  return true;
816 }

◆ CloseFile()

bool XrdThrottleManager::CloseFile ( const std::string &  entity)

Definition at line 279 of file XrdThrottleManager.cc.

280 {
281  if (m_max_open == 0 && m_max_conns == 0) return true;
282 
283  bool result = true;
284  const std::lock_guard<std::mutex> lock(m_file_mutex);
285  if (m_max_open) {
286  auto iter = m_file_counters.find(entity);
287  if (iter == m_file_counters.end()) {
288  TRACE(FILES, "WARNING: User " << entity << " closed a file but throttle plugin never saw an open file");
289  result = false;
290  } else if (iter->second == 0) {
291  TRACE(FILES, "WARNING: User " << entity << " closed a file but throttle plugin thinks all files were already closed");
292  result = false;
293  } else {
294  iter->second--;
295  }
296  if (result) TRACE(FILES, "User " << entity << " closed a file; " << iter->second <<
297  " remain open");
298  }
299 
300  if (m_max_conns) {
301  auto pid = XrdSysThread::Num();
302  auto conn_iter = m_active_conns.find(entity);
303  auto conn_count_iter = m_conn_counters.find(entity);
304  if (conn_iter == m_active_conns.end() || !(conn_iter->second)) {
305  TRACE(CONNS, "WARNING: User " << entity << " closed a file on a connection we are not"
306  " tracking");
307  return false;
308  }
309  auto pid_iter = conn_iter->second->find(pid);
310  if (pid_iter == conn_iter->second->end()) {
311  TRACE(CONNS, "WARNING: User " << entity << " closed a file on a connection we are not"
312  " tracking");
313  return false;
314  }
315  if (pid_iter->second == 0) {
316  TRACE(CONNS, "WARNING: User " << entity << " closed a file on connection the throttle"
317  " plugin thinks was idle");
318  } else {
319  pid_iter->second--;
320  }
321  if (conn_count_iter == m_conn_counters.end()) {
322  TRACE(CONNS, "WARNING: User " << entity << " closed a file but the throttle plugin never"
323  " observed an open file");
324  } else if (pid_iter->second == 0) {
325  if (conn_count_iter->second == 0) {
326  TRACE(CONNS, "WARNING: User " << entity << " had a connection go idle but the "
327  " throttle plugin already thought all connections were idle");
328  } else {
329  conn_count_iter->second--;
330  TRACE(CONNS, "User " << entity << " had connection on thread " << pid << " go idle; "
331  << conn_count_iter->second << " active connections remain");
332  }
333  }
334  }
335 
336  return result;
337 }
static unsigned long Num(void)

References XrdSysThread::Num(), and TRACE.

+ Here is the call graph for this function:

◆ FromConfig()

void XrdThrottleManager::FromConfig ( XrdThrottle::Configuration &  config)

Definition at line 64 of file XrdThrottleManager.cc.

65 {
66 
67  auto max_open = config.GetMaxOpen();
68  if (max_open != -1) SetMaxOpen(max_open);
69  auto max_conn = config.GetMaxConn();
70  if (max_conn != -1) SetMaxConns(max_conn);
71  auto max_wait = config.GetMaxWait();
72  if (max_wait != -1) SetMaxWait(max_wait);
73 
75  config.GetThrottleIOPSRate(),
76  config.GetThrottleConcurrency(),
77  static_cast<float>(config.GetThrottleRecomputeIntervalMS())/1000.0);
78 
79  m_trace->What = config.GetTraceLevels();
80 
81  auto loadshed_host = config.GetLoadshedHost();
82  auto loadshed_port = config.GetLoadshedPort();
83  auto loadshed_freq = config.GetLoadshedFreq();
84  if (!loadshed_host.empty() && loadshed_port > 0 && loadshed_freq > 0)
85  {
86  // Loadshed specified, so set it.
87  SetLoadShed(loadshed_host, loadshed_port, loadshed_freq);
88  }
89 
90  // Load per-user configuration if specified
91  auto user_config_file = config.GetUserConfigFile();
92  if (!user_config_file.empty())
93  {
94  m_user_config_file = user_config_file;
95  if (LoadUserLimits(user_config_file) != 0)
96  {
97  m_log->Emsg("ThrottleManager", "Failed to load per-user configuration file", user_config_file.c_str());
98  }
99  }
100 }
int Emsg(const char *esfx, int ecode, const char *text1, const char *text2=0)
Definition: XrdSysError.cc:116
void SetThrottles(float reqbyterate, float reqoprate, int concurrency, float interval_length)
void SetMaxOpen(unsigned long max_open)
void SetLoadShed(std::string &hostname, unsigned port, unsigned frequency)
int LoadUserLimits(const std::string &config_file)
void SetMaxWait(unsigned long max_wait)
void SetMaxConns(unsigned long max_conns)
long long GetLoadshedPort() const
long long GetThrottleDataRate() const
long long GetThrottleConcurrency() const
const std::string & GetLoadshedHost() const
long long GetLoadshedFreq() const
long long GetThrottleIOPSRate() const
long long GetThrottleRecomputeIntervalMS() const
const std::string & GetUserConfigFile() const

References XrdSysError::Emsg(), XrdThrottle::Configuration::GetLoadshedFreq(), XrdThrottle::Configuration::GetLoadshedHost(), XrdThrottle::Configuration::GetLoadshedPort(), XrdThrottle::Configuration::GetMaxConn(), XrdThrottle::Configuration::GetMaxOpen(), XrdThrottle::Configuration::GetMaxWait(), XrdThrottle::Configuration::GetThrottleConcurrency(), XrdThrottle::Configuration::GetThrottleDataRate(), XrdThrottle::Configuration::GetThrottleIOPSRate(), XrdThrottle::Configuration::GetThrottleRecomputeIntervalMS(), XrdThrottle::Configuration::GetTraceLevels(), XrdThrottle::Configuration::GetUserConfigFile(), LoadUserLimits(), SetLoadShed(), SetMaxConns(), SetMaxOpen(), SetMaxWait(), SetThrottles(), and XrdOucTrace::What.

+ Here is the call graph for this function:

◆ GetUserInfo()

std::tuple< std::string, uint16_t > XrdThrottleManager::GetUserInfo ( const XrdSecEntity *  client)

Definition at line 132 of file XrdThrottleManager.cc.

132  {
133  // client can be null, if so, return nobody
134  if (!client) {
135  return std::make_tuple("nobody", GetUid("nobody"));
136  }
137 
138  // Try various potential "names" associated with the request, from the most
139  // specific to most generic.
140  std::string user;
141 
142  if (client->eaAPI && client->eaAPI->Get("token.subject", user)) {
143  if (client->vorg) user = std::string(client->vorg) + ":" + user;
144  } else if (client->eaAPI) {
145  std::string request_name;
146  if (client->eaAPI->Get("request.name", request_name) && !request_name.empty()) user = request_name;
147  }
148  if (user.empty()) {user = client->name ? client->name : "nobody";}
149  uint16_t uid = GetUid(user.c_str());
150  return std::make_tuple(user, uid);
151 }
XrdSecAttr * Get(const void *sigkey)
char * vorg
Entity's virtual organization(s)
Definition: XrdSecEntity.hh:71
XrdSecEntityAttr * eaAPI
non-const API to attributes
Definition: XrdSecEntity.hh:92
char * name
Entity's name.
Definition: XrdSecEntity.hh:69

References XrdSecEntity::eaAPI, XrdSecEntityAttr::Get(), XrdSecEntity::name, and XrdSecEntity::vorg.

+ Here is the call graph for this function:

◆ GetUserMaxConn()

unsigned long XrdThrottleManager::GetUserMaxConn ( const std::string &  username)

Definition at line 960 of file XrdThrottleManager.cc.

961 {
962  std::shared_lock lock(m_user_limits_mutex);
963 
964  // First, try exact match
965  auto exact_iter = m_user_limits.find(username);
966  if (exact_iter != m_user_limits.end() && !exact_iter->second.is_wildcard)
967  {
968  return exact_iter->second.max_conn;
969  }
970 
971  // Then, try wildcard matches (prefer longest matching prefix)
972  unsigned long best_match = 0;
973  size_t best_prefix_len = 0;
974  unsigned long catch_all_match = 0;
975 
976  for (const auto &entry : m_user_limits)
977  {
978  if (!entry.second.is_wildcard) continue;
979 
980  const std::string &pattern = entry.first;
981 
982  // Special case: "*" is a catch-all pattern - store it but don't use it yet
983  if (pattern == "*")
984  {
985  catch_all_match = entry.second.max_conn;
986  continue;
987  }
988 
989  size_t wildcard_pos = pattern.find('*');
990  if (wildcard_pos == std::string::npos) continue;
991 
992  // Extract prefix before wildcard
993  std::string prefix = pattern.substr(0, wildcard_pos);
994  if (username.length() >= prefix.length() &&
995  username.substr(0, prefix.length()) == prefix)
996  {
997  // Prefer longer prefix matches
998  if (prefix.length() > best_prefix_len)
999  {
1000  best_prefix_len = prefix.length();
1001  best_match = entry.second.max_conn;
1002  }
1003  }
1004  }
1005 
1006  // If we found a specific wildcard match, use it
1007  if (best_match > 0) return best_match;
1008 
1009  // If no specific wildcard match, use catch-all if available
1010  // return 0 if not
1011  return catch_all_match;
1012 
1013 }

Referenced by OpenFile().

+ Here is the caller graph for this function:

◆ Init()

void XrdThrottleManager::Init ( )

Definition at line 103 of file XrdThrottleManager.cc.

104 {
105  TRACE(DEBUG, "Initializing the throttle manager.");
106  // Initialize all our shares to zero.
107  m_primary_bytes_shares.resize(m_max_users);
108  m_secondary_bytes_shares.resize(m_max_users);
109  m_primary_ops_shares.resize(m_max_users);
110  m_secondary_ops_shares.resize(m_max_users);
111  for (auto & waiter : m_waiter_info) {
112  waiter.m_manager = this;
113  }
114 
115  // Allocate each user 100KB and 10 ops to bootstrap;
116  for (int i=0; i<m_max_users; i++)
117  {
118  m_primary_bytes_shares[i] = m_last_round_allocation;
119  m_secondary_bytes_shares[i] = 0;
120  m_primary_ops_shares[i] = 10;
121  m_secondary_ops_shares[i] = 0;
122  }
123 
124  int rc;
125  pthread_t tid;
126  if ((rc = XrdSysThread::Run(&tid, XrdThrottleManager::RecomputeBootstrap, static_cast<void *>(this), 0, "Buffer Manager throttle")))
127  m_log->Emsg("ThrottleManager", rc, "create throttle thread");
128 
129 }
#define DEBUG(x)
Definition: XrdBwmTrace.hh:54
static int Run(pthread_t *, void *(*proc)(void *), void *arg, int opts=0, const char *desc=0)

References DEBUG, XrdSysError::Emsg(), XrdSysThread::Run(), and TRACE.

+ Here is the call graph for this function:

◆ IsThrottling()

bool XrdThrottleManager::IsThrottling ( )
inline

Definition at line 70 of file XrdThrottleManager.hh.

70 {return (m_ops_per_second > 0) || (m_bytes_per_second > 0);}

◆ LoadUserLimits()

int XrdThrottleManager::LoadUserLimits ( const std::string &  config_file)

Definition at line 884 of file XrdThrottleManager.cc.

885 {
886  INIReader reader(config_file);
887  if (reader.ParseError() < 0)
888  {
889  m_log->Emsg("ThrottleManager", errno, "Unable to open per-user configuration file", config_file.c_str());
890  return 1;
891  }
892  else if (reader.ParseError() > 0)
893  {
894  std::stringstream ss;
895  ss << "Parse error on line " << reader.ParseError() << " of file " << config_file;
896  m_log->Emsg("ThrottleManager", ss.str().c_str());
897  return 1;
898  }
899 
900  std::unordered_map<std::string, UserLimit> new_limits;
901 
902  // Process all sections
903  for (const auto &section : reader.Sections())
904  {
905  // Get the name parameter (required for all sections)
906  std::string name = reader.Get(section, "name", "");
907  if (name.empty())
908  {
909  m_log->Say("ThrottleManager", "Section", section.c_str(), "missing 'name' parameter; skipping");
910  continue;
911  }
912 
913  long max_conn = reader.GetInteger(section, "maxconn", 0);
914  if (max_conn <= 0)
915  {
916  m_log->Say("ThrottleManager", "Section", section.c_str(), "has invalid or missing 'maxconn' parameter; skipping");
917  continue;
918  }
919 
920  UserLimit limit;
921  limit.max_conn = static_cast<unsigned long>(max_conn);
922  // Check if name contains wildcard (including '*' for default/catch-all)
923  limit.is_wildcard = (name.find('*') != std::string::npos);
924  new_limits[name] = limit;
925  }
926 
927  // Atomically replace the limits map
928  size_t num_entries = new_limits.size();
929  {
930  std::unique_lock<std::shared_mutex> lock(m_user_limits_mutex);
931  m_user_limits = std::move(new_limits);
932  }
933 
934  m_log->Say("ThrottleManager", "Loaded", std::to_string(num_entries).c_str(), "per-user limit entries from", config_file.c_str());
935  return 0;
936 }
void Say(const char *text1, const char *text2=0, const char *txt3=0, const char *text4=0, const char *text5=0, const char *txt6=0)
Definition: XrdSysError.cc:162

References XrdSysError::Emsg(), and XrdSysError::Say().

Referenced by FromConfig(), and ReloadUserLimits().

+ Here is the call graph for this function:
+ Here is the caller graph for this function:

◆ OpenFile()

bool XrdThrottleManager::OpenFile ( const std::string &  entity,
std::string &  open_error_message 
)

Definition at line 195 of file XrdThrottleManager.cc.

196 {
197  // Get per-user connection limit (0 means use global)
198  unsigned long user_max_conn = GetUserMaxConn(entity);
199  unsigned long effective_max_conn = user_max_conn < m_max_conns ? user_max_conn : m_max_conns;
200 
201  if (m_max_open == 0 && effective_max_conn == 0) return true;
202 
203  const std::lock_guard<std::mutex> lock(m_file_mutex);
204  auto iter = m_file_counters.find(entity);
205  unsigned long cur_open_files = 0, cur_open_conns;
206  if (m_max_open) {
207  if (iter == m_file_counters.end()) {
208  m_file_counters[entity] = 1;
209  TRACE(FILES, "User " << entity << " has opened their first file");
210  cur_open_files = 1;
211  } else if (iter->second < m_max_open) {
212  iter->second++;
213  cur_open_files = iter->second;
214  } else {
215  std::stringstream ss;
216  ss << "User " << entity << " has hit the limit of " << m_max_open << " open files";
217  TRACE(FILES, ss.str());
218  error_message = ss.str();
219  return false;
220  }
221  }
222 
223  if (effective_max_conn > 0) {
224  auto pid = XrdSysThread::Num();
225  auto conn_iter = m_active_conns.find(entity);
226  auto conn_count_iter = m_conn_counters.find(entity);
227  if ((conn_count_iter != m_conn_counters.end()) && (conn_count_iter->second == effective_max_conn) &&
228  (conn_iter == m_active_conns.end() || ((*(conn_iter->second))[pid] == 0)))
229  {
230  // note: we are rolling back the increment in open files
231  if (m_max_open) iter->second--;
232  std::stringstream ss;
233  ss << "User " << entity << " has hit the limit of " << effective_max_conn <<
234  " open connections";
235  if (user_max_conn > 0) {
236  ss << " (per-user limit)";
237  }
238  TRACE(CONNS, ss.str());
239  error_message = ss.str();
240  return false;
241  }
242  if (conn_iter == m_active_conns.end()) {
243  std::unique_ptr<std::unordered_map<pid_t, unsigned long>> conn_map(
244  new std::unordered_map<pid_t, unsigned long>());
245  (*conn_map)[pid] = 1;
246  m_active_conns[entity] = std::move(conn_map);
247  if (conn_count_iter == m_conn_counters.end()) {
248  m_conn_counters[entity] = 1;
249  cur_open_conns = 1;
250  } else {
251  m_conn_counters[entity] ++;
252  cur_open_conns = m_conn_counters[entity];
253  }
254  } else {
255  auto pid_iter = conn_iter->second->find(pid);
256  if (pid_iter == conn_iter->second->end() || pid_iter->second == 0) {
257  (*(conn_iter->second))[pid] = 1;
258  conn_count_iter->second++;
259  cur_open_conns = conn_count_iter->second;
260  } else {
261  (*(conn_iter->second))[pid] ++;
262  cur_open_conns = conn_count_iter->second;
263  }
264  }
265  TRACE(CONNS, "User " << entity << " has " << cur_open_conns << " open connections (limit: " << effective_max_conn << ")");
266  }
267  if (m_max_open) TRACE(FILES, "User " << entity << " has " << cur_open_files << " open files");
268  return true;
269 }
unsigned long GetUserMaxConn(const std::string &username)

References GetUserMaxConn(), XrdSysThread::Num(), and TRACE.

+ Here is the call graph for this function:

◆ PerformLoadShed()

void XrdThrottleManager::PerformLoadShed ( const std::string &  opaque,
std::string &  host,
unsigned &  port 
)

Definition at line 843 of file XrdThrottleManager.cc.

844 {
845  host = m_loadshed_host;
846  host += "?";
847  host += opaque;
848  port = m_loadshed_port;
849 }

◆ PrepLoadShed()

void XrdThrottleManager::PrepLoadShed ( const char *  opaque,
std::string &  lsOpaque 
)

Definition at line 819 of file XrdThrottleManager.cc.

820 {
821  if (m_loadshed_port == 0)
822  {
823  return;
824  }
825  if (opaque && opaque[0])
826  {
827  XrdOucEnv env(opaque);
828  // Do not load shed client if it has already been done once.
829  if (env.Get("throttle.shed") != 0)
830  {
831  return;
832  }
833  lsOpaque = opaque;
834  lsOpaque += "&throttle.shed=1";
835  }
836  else
837  {
838  lsOpaque = "throttle.shed=1";
839  }
840 }

References XrdOucEnv::Get().

+ Here is the call graph for this function:

◆ ReloadUserLimits()

int XrdThrottleManager::ReloadUserLimits ( )

Definition at line 942 of file XrdThrottleManager.cc.

943 {
944  if (m_user_config_file.empty())
945  {
946  m_log->Emsg("ThrottleManager", "No per-user configuration file specified");
947  return 1;
948  }
949  return LoadUserLimits(m_user_config_file);
950 }

References XrdSysError::Emsg(), and LoadUserLimits().

+ Here is the call graph for this function:

◆ SetLoadShed()

void XrdThrottleManager::SetLoadShed ( std::string &  hostname,
unsigned  port,
unsigned  frequency 
)
inline

Definition at line 81 of file XrdThrottleManager.hh.

82  {m_loadshed_host = hostname; m_loadshed_port = port; m_loadshed_frequency = frequency;}

Referenced by FromConfig().

+ Here is the caller graph for this function:

◆ SetMaxConns()

void XrdThrottleManager::SetMaxConns ( unsigned long  max_conns)
inline

Definition at line 86 of file XrdThrottleManager.hh.

86 {m_max_conns = max_conns;}

Referenced by FromConfig().

+ Here is the caller graph for this function:

◆ SetMaxOpen()

void XrdThrottleManager::SetMaxOpen ( unsigned long  max_open)
inline

Definition at line 84 of file XrdThrottleManager.hh.

84 {m_max_open = max_open;}

Referenced by FromConfig().

+ Here is the caller graph for this function:

◆ SetMaxWait()

void XrdThrottleManager::SetMaxWait ( unsigned long  max_wait)
inline

Definition at line 88 of file XrdThrottleManager.hh.

88 {m_max_wait_time = std::chrono::seconds(max_wait);}

Referenced by FromConfig().

+ Here is the caller graph for this function:

◆ SetMonitor()

void XrdThrottleManager::SetMonitor ( XrdXrootdGStream *  gstream)
inline

Definition at line 101 of file XrdThrottleManager.hh.

101 {m_gstream = gstream;}

◆ SetThrottles()

void XrdThrottleManager::SetThrottles ( float  reqbyterate,
float  reqoprate,
int  concurrency,
float  interval_length 
)
inline

Definition at line 77 of file XrdThrottleManager.hh.

78  {m_interval_length_seconds = interval_length; m_bytes_per_second = reqbyterate;
79  m_ops_per_second = reqoprate; m_concurrency_limit = concurrency;}

Referenced by FromConfig().

+ Here is the caller graph for this function:

◆ StartIOTimer()

XrdThrottleTimer XrdThrottleManager::StartIOTimer ( uint16_t  uid,
bool &  ok 
)

Definition at line 731 of file XrdThrottleManager.cc.

732 {
733  int cur_counter = m_io_active.fetch_add(1, std::memory_order_acq_rel);
734  m_io_total++;
735 
736  while (m_concurrency_limit >= 0 && cur_counter >= m_concurrency_limit)
737  {
738  // If the user has essentially no concurrency, then we let them
739  // temporarily exceed the limit. This prevents potential waits for
740  // every single read for an infrequent user.
741  if (m_waiter_info[uid].m_concurrency < 1)
742  {
743  break;
744  }
745  m_loadshed_limit_hit++;
746  m_io_active.fetch_sub(1, std::memory_order_acq_rel);
747  TRACE(DEBUG, "ThrottleManager (user=" << uid << "): IO concurrency limit hit; waiting for other IOs to finish.");
748  ok = m_waiter_info[uid].Wait();
749  if (!ok) {
750  TRACE(DEBUG, "ThrottleManager (user=" << uid << "): timed out waiting for other IOs to finish.");
751  return XrdThrottleTimer();
752  }
753  cur_counter = m_io_active.fetch_add(1, std::memory_order_acq_rel);
754  }
755 
756  ok = true;
757  return XrdThrottleTimer(this, uid);
758 }
friend class XrdThrottleTimer

References DEBUG, TRACE, and XrdThrottleTimer.

◆ StopIOTimer()

void XrdThrottleManager::StopIOTimer ( std::chrono::steady_clock::duration &  event_duration,
uint16_t  uid 
)
protected

Definition at line 764 of file XrdThrottleManager.cc.

765 {
766  m_io_active_time += event_duration.count();
767  auto old_active = m_io_active.fetch_sub(1, std::memory_order_acq_rel);
768  m_waiter_info[uid].m_io_time += event_duration.count();
769  if (old_active == static_cast<unsigned>(m_concurrency_limit))
770  {
771  // If we are below the concurrency limit threshold and have another waiter
772  // for our user, then execute it immediately. Otherwise, we will give
773  // someone else a chance to run (as we have gotten more than our share recently).
774  unsigned waiting_users = m_waiting_users;
775  if (waiting_users == 0) waiting_users = 1;
776  if (m_waiter_info[uid].m_concurrency < m_concurrency_limit / waiting_users)
777  {
778  std::unique_lock<std::mutex> lock(m_waiter_info[uid].m_mutex);
779  if (m_waiter_info[uid].m_waiting > 0)
780  {
781  m_waiter_info[uid].NotifyOne(std::move(lock));
782  return;
783  }
784  }
785  NotifyOne();
786  }
787 }

Friends And Related Function Documentation

◆ XrdThrottleTimer

friend class XrdThrottleTimer
friend

Definition at line 57 of file XrdThrottleManager.hh.

Referenced by StartIOTimer().


The documentation for this class was generated from the following files: