GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/quota_listener.cc
Date: 2025-11-09 02:35:23
Exec Total Coverage
Lines: 0 76 0.0%
Branches: 0 42 0.0%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5
6 #include "quota_listener.h"
7
8 #include <poll.h>
9 #include <pthread.h>
10
11 #include <cstdlib>
12
13 #include "catalog_mgr.h"
14 #include "quota.h"
15 #include "util/exception.h"
16 #include "util/logging.h"
17 #include "util/posix.h"
18 #include "util/smalloc.h"
19
20 using namespace std; // NOLINT
21
22 namespace quota {
23
24 struct ListenerHandle {
25 int pipe_backchannel[2];
26 int pipe_terminate[2];
27 QuotaManager *quota_manager;
28 CatalogManager *catalog_manager;
29 string repository_name;
30 pthread_t thread_listener;
31 };
32
33
34 ✗ static void *MainUnpinListener(void *data) {
35 ✗ ListenerHandle *handle = static_cast<ListenerHandle *>(data);
36 ✗ LogCvmfs(kLogQuota, kLogDebug, "starting unpin listener for %s",
37 handle->repository_name.c_str());
38
39 struct pollfd *watch_fds = static_cast<struct pollfd *>(
40 ✗ smalloc(2 * sizeof(struct pollfd)));
41 ✗ watch_fds[0].fd = handle->pipe_terminate[0];
42 ✗ watch_fds[0].events = POLLIN | POLLPRI;
43 ✗ watch_fds[0].revents = 0;
44 ✗ watch_fds[1].fd = handle->pipe_backchannel[0];
45 ✗ watch_fds[1].events = POLLIN | POLLPRI;
46 ✗ watch_fds[1].revents = 0;
47 while (true) {
48 ✗ const int retval = poll(watch_fds, 2, -1);
49 ✗ if (retval < 0) {
50 ✗ continue;
51 }
52
53 // Terminate I/O thread
54 ✗ if (watch_fds[0].revents)
55 ✗ break;
56
57 // Release pinned catalogs
58 ✗ if (watch_fds[1].revents) {
59 ✗ watch_fds[1].revents = 0;
60 char cmd;
61 ✗ ReadPipe(handle->pipe_backchannel[0], &cmd, sizeof(cmd));
62 ✗ if (cmd == 'R') {
63 ✗ handle->catalog_manager->DetachNested();
64 ✗ LogCvmfs(kLogQuota, kLogDebug | kLogSyslog, "released nested catalogs");
65 }
66 }
67 }
68 ✗ free(watch_fds);
69
70 ✗ LogCvmfs(kLogQuota, kLogDebug, "stopping unpin listener for %s",
71 handle->repository_name.c_str());
72 ✗ return NULL;
73 }
74
75
76 ✗ static void *MainWatchdogListener(void *data) {
77 ✗ ListenerHandle *handle = static_cast<ListenerHandle *>(data);
78 ✗ LogCvmfs(kLogQuota, kLogDebug, "starting cache manager watchdog for %s",
79 handle->repository_name.c_str());
80
81 struct pollfd *watch_fds = static_cast<struct pollfd *>(
82 ✗ smalloc(2 * sizeof(struct pollfd)));
83 ✗ watch_fds[0].fd = handle->pipe_terminate[0];
84 ✗ watch_fds[0].events = POLLIN | POLLPRI;
85 ✗ watch_fds[0].revents = 0;
86 ✗ watch_fds[1].fd = handle->pipe_backchannel[0];
87 ✗ watch_fds[1].events = POLLIN | POLLPRI;
88 ✗ watch_fds[1].revents = 0;
89 while (true) {
90 ✗ const int retval = poll(watch_fds, 2, -1);
91 ✗ if (retval < 0) {
92 ✗ continue;
93 }
94
95 // Terminate I/O thread
96 ✗ if (watch_fds[0].revents)
97 ✗ break;
98
99 // Release pinned catalogs
100 ✗ if (watch_fds[1].revents) {
101 ✗ if ((watch_fds[1].revents & POLLERR) || (watch_fds[1].revents & POLLHUP)
102 ✗ || (watch_fds[1].revents & POLLNVAL)) {
103 ✗ PANIC(kLogDebug | kLogSyslogErr, "cache manager disappeared, aborting");
104 }
105 // Clean the pipe
106 ✗ watch_fds[1].revents = 0;
107 char dummy;
108 ✗ ReadPipe(handle->pipe_backchannel[0], &dummy, sizeof(dummy));
109 }
110 }
111 ✗ free(watch_fds);
112
113 ✗ LogCvmfs(kLogQuota, kLogDebug, "stopping cache manager watchdog for %s",
114 handle->repository_name.c_str());
115 ✗ return NULL;
116 }
117
118
119 /**
120 * Registers a back channel that reacts on high watermark of pinned chunks
121 */
122 ✗ ListenerHandle *RegisterUnpinListener(QuotaManager *quota_manager,
123 CatalogManager *catalog_manager,
124 const string &repository_name) {
125 ✗ ListenerHandle *handle = new ListenerHandle();
126 ✗ quota_manager->RegisterBackChannel(handle->pipe_backchannel, repository_name);
127 ✗ MakePipe(handle->pipe_terminate);
128 ✗ handle->quota_manager = quota_manager;
129 ✗ handle->catalog_manager = catalog_manager;
130 ✗ handle->repository_name = repository_name;
131 ✗ const int retval = pthread_create(&handle->thread_listener, NULL,
132 MainUnpinListener,
133 static_cast<void *>(handle));
134 ✗ assert(retval == 0);
135 ✗ return handle;
136 }
137
138
139 /**
140 * Registers a back channel that kills the fuse module if the cache manager
141 * disappears
142 */
143 ✗ ListenerHandle *RegisterWatchdogListener(QuotaManager *quota_manager,
144 const string &repository_name) {
145 ✗ ListenerHandle *handle = new ListenerHandle();
146 ✗ quota_manager->RegisterBackChannel(handle->pipe_backchannel, repository_name);
147 ✗ MakePipe(handle->pipe_terminate);
148 ✗ handle->quota_manager = quota_manager;
149 ✗ handle->catalog_manager = NULL;
150 ✗ handle->repository_name = repository_name;
151 ✗ const int retval = pthread_create(&handle->thread_listener, NULL,
152 MainWatchdogListener,
153 static_cast<void *>(handle));
154 ✗ assert(retval == 0);
155 ✗ return handle;
156 }
157
158
159 ✗ void UnregisterListener(ListenerHandle *handle) {
160 ✗ const char terminate = 'T';
161 ✗ WritePipe(handle->pipe_terminate[1], &terminate, sizeof(terminate));
162 ✗ pthread_join(handle->thread_listener, NULL);
163 ✗ ClosePipe(handle->pipe_terminate);
164
165 ✗ handle->quota_manager->UnregisterBackChannel(handle->pipe_backchannel,
166 ✗ handle->repository_name);
167
168 ✗ delete handle;
169 }
170
171 } // namespace quota
172