error: ‘this’ was not captured for this lambda function

[edwardcapriolo]

Hello all,
I am a true nube (to seastar and c++) so please bare with me in some RTFM/you should know that type questions

I am trying to follow the example here:

https://seastar.io/message-passing/

[1] return conn->read_exactly(4).then(temporary_buffer buf) {
[2] int id = buf_to_id(buf);
[3] return smp::submit_to(other_core, [id] {
[4] return lookup(id);
[5] });
[6] }).then([this] (sstring result) {
[7] return conn->write(result);
[8] });

I was going to kick the tires on a small "distributed kv strore"

The sharded seemed nice so I am am trying to follow the advice.

#include <iostream>
#include "config.h"

#include <seastar/core/app-template.hh>
#include <seastar/core/reactor.hh>
#include <seastar/core/sleep.hh>
#include <seastar/core/seastar.hh>
#include <seastar/core/reactor.hh>
#include <seastar/core/future-util.hh>
#include <seastar/net/api.hh>
#include <iostream>
#include <seastar/core/sharded.hh>
#include <unordered_set>

struct my_datum {
	std::string something;
	bool operator==(const my_datum&) const = default;
};

template <>
struct std::hash<my_datum> {
	std::size_t operator()(const my_datum& my_datum) const noexcept {
		return std::hash<std::string>{}(my_datum.something);
	}
};


class my_service {
	public:
		std::unordered_set<my_datum> data;
		my_service(){

		}

		seastar::future<> run() {
			std::cerr << "running on " << seastar::this_shard_id() << "\n";
			return seastar::make_ready_future<>();
		}

		seastar::future<> insert(my_datum& my_datum){
			std:size_t hash = std::hash<std::string>{}(my_datum.something);
			int ct = seastar::smp::count;
			int core = ct % hash;
			return seastar::smp::submit_to(core, [my_datum] {
				local_insert(my_datum);
			});
		}
		void local_insert(my_datum& my_datum){
			data.insert(my_datum);
		}

		int find_shard(int data){
			int ct = seastar::smp::count;
			return data % ct;
		}

		seastar::future<> stop(){
			return seastar::make_ready_future<>();
		}
};

seastar::sharded<my_service> service;


seastar::future<> k() {
  return seastar::make_ready_future<>();
}

int main(int argc, char** argv) {
    seastar::app_template app;
    try {
        app.run(argc, argv, k);
    } catch(...) {
        std::cerr << "Couldn't start application: "
                  << std::current_exception() << "\n";
        return 1;
    }
    return 0;
}

The error I get looks like this:

AILED: CMakeFiles/abc.dir/abc.cc.o
/usr/bin/c++ -DFMT_SHARED -DSEASTAR_API_LEVEL=7 -DSEASTAR_BROKEN_SOURCE_LOCATION -DSEASTAR_DEPRECATED_OSTREAM_FORMATTERS -DSEASTAR_LOGGER_COMPILE_TIME_FMT -DSEASTAR_LOGGER_TYPE_STDOUT -DSEASTAR_P2581R1 -DSEASTAR_SCHEDULING_GROUPS_COUNT=16 -DSEASTAR_SSTRING -I/home/edward/eclipse-workspace/abc/build/cmake.run.linux.x86_64.Local -isystem /home/edward/seastar/include -isystem /home/edward/seastar/build/release/gen/include -isystem /home/edward/seastar/build/release/gen/src -O3 -DNDEBUG -std=gnu++23 -Wno-maybe-uninitialized -Wno-error=unused-result -MD -MT CMakeFiles/abc.dir/abc.cc.o -MF CMakeFiles/abc.dir/abc.cc.o.d -o CMakeFiles/abc.dir/abc.cc.o -c /home/edward/eclipse-workspace/abc/abc.cc
/home/edward/eclipse-workspace/abc/abc.cc: In lambda function:
/home/edward/eclipse-workspace/abc/abc.cc:45:45: error: ‘this’ was not captured for this lambda function
45 | local_insert(my_datum);
| ~~~~~~~~~~~~^~~~~~~~~~
/home/edward/eclipse-workspace/abc/abc.cc:45:45: error: cannot call member function ‘void my_service::local_insert(my_datum&)’ without object

It is not entirely clear to me how to design this. If look is static method, it wont easiliy have access to

std::unordered_set<my_datum> data;

I also am hoping to avoid any pattern where each private method has a companion.
Thanks

[avikivity]

Use sharded::invoke_on to receive a reference to the local service as a parameter.

[edwardcapriolo]

I have reworked a few examples together:

#include <iostream>
#include "config.h"

#include <seastar/core/app-template.hh>
#include <seastar/core/reactor.hh>
#include <seastar/core/sleep.hh>
#include <seastar/core/seastar.hh>
#include <seastar/core/reactor.hh>
#include <seastar/core/future-util.hh>
#include <seastar/net/api.hh>
#include <iostream>
#include <seastar/core/sharded.hh>
#include <unordered_set>
#include <seastar/core/distributed.hh>

struct my_datum {
	std::string something;
	bool operator==(const my_datum&) const = default;
};

template <>
struct std::hash<my_datum> {
	std::size_t operator()(const my_datum& my_datum) const noexcept {
		return std::hash<std::string>{}(my_datum.something);
	}
};

struct local_origin_tag {
    template <typename T>
    static inline
    T move_if_local(T& ref) {
        return std::move(ref);
    }
};

//https://github.com/scylladb/seastar/blob/master/apps/memcached/memcache.cc#L347C1-L353C3
//i dont quite understand this
struct remote_origin_tag {
    template <typename T>
    static inline
    T move_if_local(T& ref) {
        return ref;
    }
};

class cache {
private:
	std::unordered_set<my_datum> data;
public:
	template <typename Origin = local_origin_tag>
	bool insert(my_datum& my_datum){
		data.insert(my_datum);
		return true;
	}
};

class sharded_cache {
private:
	seastar::distributed<cache>& _peers;
	inline
	unsigned get_shard(const my_datum& data){
		return std::hash<my_datum>{}(data) % seastar::smp::count;
	}

public:
	sharded_cache(seastar::distributed<cache>& peers) : _peers(peers) {

	}
	seastar::future<bool> insert(my_datum& my_datum){
		auto cpu = get_shard(my_datum);
		if (seastar::this_shard_id() == cpu){
			return seastar::make_ready_future<bool>( _peers.local().insert(my_datum));
		} else {
			return _peers.invoke_on(cpu, &cache::insert<local_origin_tag>, std::ref(my_datum)  );
		}
	}
};

class tcp_server {
public:
	tcp_server(sharded_cache& cache) : _cache(cache){

	}
	seastar::future<> init() {
	    seastar::listen_options lo;
	    lo.reuse_address = true;
	    return seastar::do_with(
	    		seastar::listen(seastar::make_ipv4_address({1234}), lo),
				this,
	            [] (auto& listener, auto& me) {
	        return seastar::keep_doing([&listener, &me] () {
	            return listener.accept().then(
	                    [&] (seastar::accept_result res) {
	            	(void) me->handle_connection(std::move(res.connection)).handle_exception(
	    	                        [] (std::exception_ptr ep) {
	    	                    fmt::print(stderr, "Could not handle connection: {}\n", ep);
	    	                });
	            });
	        });
	    });
	}
private:

	seastar::future<> handle_connection(seastar::connected_socket s) {
	    auto out = s.output();
	    auto in = s.input();
	    return do_with(std::move(s), std::move(out), std::move(in), this,
	            [] (auto& s, auto& out, auto& in, auto& me) {
	        return seastar::repeat([&out, &in, &me] {
	            return in.read().then([&out, &me] (auto buf) {
	                if (buf) {
	                    return out.write(std::move(buf)).then([&out, &me] {
	                    	std::cerr << me;
	                    	(void) seastar::do_with(my_datum(), [&me] (auto& obj)
	                    			{
	                    				obj.something = "yo";
	                    				return me->_cache.insert(obj);
	                    			} );
	                    	return out.flush();
	                    }).then([] {
	                        return seastar::stop_iteration::no;
	                    });
	                } else {
	                    return seastar::make_ready_future<seastar::stop_iteration>(seastar::stop_iteration::yes);
	                }
	            });
	        }).then([&out] {
	            return out.close();
	        });
	    });
	}


	sharded_cache& _cache;
};

seastar::future<> handle_connection(seastar::connected_socket s) {
    auto out = s.output();
    auto in = s.input();
    return do_with(std::move(s), std::move(out), std::move(in),
            [] (auto& s, auto& out, auto& in) {
        return seastar::repeat([&out, &in] {
            return in.read().then([&out] (auto buf) {
                if (buf) {
                    return out.write(std::move(buf)).then([&out] {
                        return out.flush();
                    }).then([] {
                        return seastar::stop_iteration::no;
                    });
                } else {
                    return seastar::make_ready_future<seastar::stop_iteration>(
                            seastar::stop_iteration::yes);
                }
            });
        }).then([&out] {
            return out.close();
        });
    });
}

int main(int argc, char** argv) {
    seastar::app_template app;
    seastar::distributed<cache> cache_peers;
    sharded_cache cache(cache_peers);
    seastar::distributed<tcp_server> tcp_server;
    try {
    	app.run(argc, argv, [&]{
    		return cache_peers.start()
    				.then([&] { return tcp_server.start(std::ref(cache)); })
					.then([&] { return tcp_server.invoke_on_all(&tcp_server::init); })
					;
		});
    } catch(...) {
        std::cerr << "Couldn't start application: "
                  << std::current_exception() << "\n";
        return 1;
    }
    return 0;
}

This was where I got most inspiration:
https://github.com/scylladb/seastar/blob/master/apps/memcached/memcache.cc

I have some questions:

(void) seastar::do_with(my_datum(), [&me] (auto& obj)
	                    			{
	                    				obj.something = "yo";
	                    				return me->_cache.insert(obj);
	                    			} );
	                    	return out.flush();

I run into this problem a few times, method signatures conflict future vs future<> is there a simple recipe for ignoring/converting the return of a future?

Also is there a better way extension vs function passing to avoid mapping & or &me into nested captures ?

[edwardcapriolo]

Yes thanks. The co-routines make sense, all the scary language in the manual about them running out of order is somewhat chasing me away.

[avikivity]

No they are wonderful.

[edwardcapriolo]

They may be "wonderful"

https://github.com/scylladb/seastar/blob/master/doc/lambda-coroutine-fiasco.md

But when your describe them as a "fiasco" in the tutorial realize that it "might" scare people off.

[avikivity]

Seastar (or just C++) isn't for people who are easily scared. Read the tutorial and understand it, then decide, don't jump to conclusions because you read a word.

[edwardcapriolo]

Seastar (or just C++) isn't for people who are easily scared. Read the tutorial and understand it, then decide, don't jump to conclusions because you read a word.

Sir, dont wake my quotes as something, insulting. Its a writing style of mine really sorry. But, I will call a spade a spade a spade "read the tutoral and understand it"

https://docs.seastar.io/master/tutorial.html

Go count up the TODOs and incomplete thoughts and realize it is not easy to understand. When you look in the code /src/test you either find examples wildly different, or the exact same code from a test copy pasted into the tutorial.

The point I was making: There is a lot to take in with when trying to absorb, cpp 23 features, cpp labdas (that arent the prettiest) coroutines, your framework, and extra rules around cooroutines and lambdas, requiring the extra thing you specify. Choose a better word than fiasco maybe.

BTW:
More information can be found in lambda-coroutine-fiasco.md.

Not a link!